paste.txt
Сущности
# 📝 SG INDEX v4.2 — ПЛАН УЛУЧШЕНИЙ P1 (ROADMAP)<br>
<br>
**Версия:** v4.2 P1 Roadmap <br>
**Период:** Февраль–Март 2026 <br>
**Статус:** 📅 **PLANNED** <br>
**Цель:** Довести качество кода и тестирование до 10/10<br>
<br>
---<br>
<br>
## 1. OVERVIEW<br>
<br>
**Зачем нужен P1:**<br>
Независимая верификация выявила 5 областей улучшения (не блокирующих launch), которые повысят надёжность и управляемость модели в долгосрочной перспективе.<br>
<br>
**Приоритет:** P1 (High) — рекомендуется до конца марта 2026 <br>
**Не блокирует:** Запуск пилота 22 января ✅ <br>
**Влияет на:** Долгосрочную стабильность, coverage, security<br>
<br>
**5 задач P1:**<br>
1. Unit-тесты (coverage → 85%)<br>
2. Edge-case тесты<br>
3. Pydantic валидация входов<br>
4. Config validator<br>
5. Anti-gaming ML<br>
<br>
---<br>
<br>
## 2. TASK 1: UNIT-ТЕСТЫ (15 ФЕВРАЛЯ)<br>
<br>
### Цель<br>
<br>
Добавить 20 unit-тестов для каждого компонента модели, чтобы:<br>
- Покрытие ≥ 85% (сейчас ~60-75%)<br>
- Упростить поиск ошибок при изменениях<br>
- Протестировать каждый шаг изолированно<br>
<br>
### Что тестировать<br>
<br>
| Компонент | Количество тестов | Примеры |<br>
|-----------|-------------------|---------|<br>
| Шаг 1: T_comp | 3 | Basic, границы [0,1], clip |<br>
| Шаг 2: S_pot (Cobb-Douglas) | 4 | Zero inputs, CRS, монотонность |<br>
| Шаг 3: F_gate | 5 | Границы (T=0, T=0.85, T=1), sigmoid overflow |<br>
| Шаг 4: F_syn | 3 | Min (C=T=0), max (C=T=1), нормализация |<br>
| Шаг 5: F_vol | 3 | σ=0, σ=50, σ>50 (clip) |<br>
| Шаг 6: S_raw (PRODUCT) | 2 | All 1.0, any 0 → result≈0 |<br>
| **TOTAL** | **20** | — |<br>
<br>
### Код примеров<br>
<br>
```python<br>
# tests/unit/test_step1_trust.py<br>
import pytest<br>
from sg_index_v42_final import SGIndexV42<br>
<br>
@pytest.fixture<br>
def model():<br>
return SGIndexV42()<br>
<br>
def test_composite_trust_basic(model):<br>
"""T_comp = 0.6*T_l + 0.4*Z"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=0, sigma=0)<br>
# При T_l=1, Z=0 → T_comp = 0.6*1 + 0.4*0 = 0.6<br>
assert 0.59 <= result.T_comp <= 0.61<br>
<br>
def test_composite_trust_full(model):<br>
"""При T_l=Z=1 → T_comp=1"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.T_comp == 1.0<br>
<br>
def test_composite_trust_clip(model):<br>
"""T_comp должно clip в [0,1]"""<br>
# Внутренний clip проверяется через граничные значения<br>
result = model.compute(C=1, V=1, T_loyalty=0, Z=0, sigma=0)<br>
assert result.T_comp == 0.0<br>
<br>
<br>
# tests/unit/test_step2_potential.py<br>
def test_cobb_douglas_zero_c(model):<br>
"""При C=0 → S_pot должен быть ≈0 (C^0.25 ≈ 0)"""<br>
result = model.compute(C=0, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.S_pot < 0.01<br>
<br>
def test_cobb_douglas_all_ones(model):<br>
"""При C=V=T=1 → S_pot=1"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert 0.99 <= result.S_pot <= 1.01<br>
<br>
def test_cobb_douglas_crs(model):<br>
"""Constant Returns to Scale: (2C, 2V, 2T) → 2*S_pot"""<br>
# Но входы ограничены [0,1], поэтому тестируем линейность внутри диапазона<br>
s1 = model.compute(C=0.5, V=0.5, T_loyalty=0.5, Z=0.5, sigma=0).S_pot<br>
s2 = model.compute(C=1.0, V=1.0, T_loyalty=1.0, Z=1.0, sigma=0).S_pot<br>
# s2/s1 должно быть около 2^1 = 2 (для CRS)<br>
ratio = s2 / s1 if s1 > 0 else 0<br>
assert 1.9 <= ratio <= 2.1 # Approximate CRS<br>
<br>
def test_cobb_douglas_monotonic_c(model):<br>
"""C↑ → S_pot↑"""<br>
s1 = model.compute(C=0.2, V=1, T_loyalty=1, Z=1, sigma=0).S_pot<br>
s2 = model.compute(C=0.8, V=1, T_loyalty=1, Z=1, sigma=0).S_pot<br>
assert s2 > s1<br>
<br>
<br>
# tests/unit/test_step3_gate.py<br>
def test_gate_at_zero(model):<br>
"""При T_comp=0 → F_gate=0"""<br>
result = model.compute(C=1, V=1, T_loyalty=0, Z=0, sigma=0)<br>
assert result.F_gate < 0.01<br>
<br>
def test_gate_at_one(model):<br>
"""При T_comp=1 → F_gate=1"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.F_gate > 0.99<br>
<br>
def test_gate_at_threshold(model):<br>
"""При T_comp=0.85 → F_gate≈0.82"""<br>
result = model.compute(C=1, V=1, T_loyalty=0.85, Z=0.85, sigma=0)<br>
assert 0.80 <= result.F_gate <= 0.85<br>
<br>
def test_gate_sigmoid_overflow_protection(model):<br>
"""Проверить, что expit защищен от overflow"""<br>
# expit должен возвращать 0/1 для очень больших/маленьких x<br>
# Косвенная проверка через граничные T<br>
result_low = model.compute(C=1, V=1, T_loyalty=0.01, Z=0.01, sigma=0)<br>
result_high = model.compute(C=1, V=1, T_loyalty=0.99, Z=0.99, sigma=0)<br>
assert 0 <= result_low.F_gate <= 1<br>
assert 0 <= result_high.F_gate <= 1<br>
<br>
def test_gate_monotonic(model):<br>
"""T↑ → F_gate↑"""<br>
s1 = model.compute(C=1, V=1, T_loyalty=0.3, Z=0.3, sigma=0).F_gate<br>
s2 = model.compute(C=1, V=1, T_loyalty=0.9, Z=0.9, sigma=0).F_gate<br>
assert s2 > s1<br>
<br>
<br>
# tests/unit/test_step4_synergy.py<br>
def test_synergy_min(model):<br>
"""При C=T=0 → F_syn=1 (нет синергии)"""<br>
result = model.compute(C=0, V=1, T_loyalty=0, Z=0, sigma=0)<br>
assert result.F_syn == 1.0<br>
<br>
def test_synergy_max(model):<br>
"""При C=T=1 → F_syn≈1.259"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert 1.25 <= result.F_syn <= 1.27<br>
<br>
def test_synergy_normalization(model):<br>
"""F_syn должна быть ограничена (не больше 1.26)"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.F_syn <= 1.26<br>
<br>
<br>
# tests/unit/test_step5_volatility.py<br>
def test_vol_zero_sigma(model):<br>
"""При σ=0 → F_vol=1 (нет штрафа)"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.F_vol == 1.0<br>
<br>
def test_vol_high_sigma(model):<br>
"""При σ=50 → F_vol=1/(1+0.1*50)=0.167"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=50)<br>
assert 0.15 <= result.F_vol <= 0.20<br>
<br>
def test_vol_clip_sigma(model):<br>
"""При σ>50 → должно clip к 50"""<br>
result1 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=50)<br>
result2 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=100)<br>
# Оба должны дать одинаковый F_vol (σ clip к 50)<br>
assert abs(result1.F_vol - result2.F_vol) < 0.01<br>
<br>
<br>
# tests/unit/test_step6_aggregation.py<br>
def test_product_all_ones(model):<br>
"""При всех компонентах=1 → S_raw=max"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
# S_pot=1, F_gate=1, F_syn≈1.259, F_vol=1 → S_raw≈1.259<br>
assert 1.25 <= result.S_raw <= 1.27<br>
<br>
def test_product_any_zero(model):<br>
"""При любом компоненте≈0 → S_raw≈0"""<br>
result = model.compute(C=0, V=1, T_loyalty=1, Z=1, sigma=0)<br>
# S_pot≈0 → S_raw≈0<br>
assert result.S_raw < 0.01<br>
```<br>
<br>
### Success Criteria<br>
<br>
- [ ] Написано 20 unit-тестов<br>
- [ ] Все тесты проходят (20/20 ✅)<br>
- [ ] Coverage ≥ 85% (pytest --cov)<br>
- [ ] CI/CD проверяет coverage автоматически<br>
<br>
### Deadline<br>
<br>
**15 февраля 2026**<br>
<br>
---<br>
<br>
## 3. TASK 2: EDGE-CASE ТЕСТЫ (20 ФЕВРАЛЯ)<br>
<br>
### Цель<br>
<br>
Протестировать граничные и аномальные случаи:<br>
- Отрицательные входы (должно clip → 0)<br>
- Очень большие входы (σ > 50, должно clip → 50)<br>
- Влияние Z отдельно от T_loyalty<br>
- Комбинации крайних значений<br>
<br>
### Что тестировать<br>
<br>
| Категория | Тесты | Цель |<br>
|-----------|-------|------|<br>
| Negative inputs | 3 | C, V, T < 0 → clip → 0 |<br>
| Overflow inputs | 3 | σ > 50 → clip → 50 |<br>
| Z influence | 2 | Отдельно от T_loyalty |<br>
| Extreme combinations | 2 | (C=0, σ=50), (C=1, σ=0) |<br>
| **TOTAL** | **10** | — |<br>
<br>
### Код примеров<br>
<br>
```python<br>
# tests/unit/test_edge_cases.py<br>
def test_negative_capacity(model):<br>
"""C < 0 должно clip к 0"""<br>
# Внутренний clip защищает, но проверим результат<br>
result = model.compute(C=-0.5, V=1, T_loyalty=1, Z=1, sigma=0)<br>
# C=0 → S_pot≈0 → S_official≈0<br>
assert result.S_official < 5<br>
<br>
def test_negative_visibility(model):<br>
"""V < 0 должно clip к 0"""<br>
result = model.compute(C=1, V=-0.5, T_loyalty=1, Z=1, sigma=0)<br>
assert result.S_official < 5<br>
<br>
def test_negative_trust(model):<br>
"""T_loyalty < 0 должно clip к 0"""<br>
result = model.compute(C=1, V=1, T_loyalty=-0.5, Z=0, sigma=0)<br>
# T_comp = 0.6*0 + 0.4*0 = 0 → низкий S<br>
assert result.S_official < 20<br>
<br>
def test_sigma_overflow(model):<br>
"""σ > 50 должно clip к 50"""<br>
result1 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=50)<br>
result2 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=100)<br>
result3 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=200)<br>
# Все должны дать один S_official (σ clip к 50)<br>
assert abs(result1.S_official - result2.S_official) < 1<br>
assert abs(result1.S_official - result3.S_official) < 1<br>
<br>
def test_sigma_negative(model):<br>
"""σ < 0 (некорректно) → clip к 0 → F_vol=1"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=-10)<br>
# σ=0 → F_vol=1 → S≈100<br>
assert result.S_official > 95<br>
<br>
def test_sigma_extreme_positive(model):<br>
"""σ=1000 (некорректно) → clip к 50"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=1000)<br>
assert 15 <= result.S_official <= 20 # F_vol≈0.167<br>
<br>
def test_z_influence_low_trust(model):<br>
"""Z влияет на T_comp отдельно от T_loyalty"""<br>
# При T_loyalty=0.5, Z=0 → T_comp=0.3<br>
s1 = model.compute(C=1, V=1, T_loyalty=0.5, Z=0, sigma=0).S_official<br>
# При T_loyalty=0.5, Z=1 → T_comp=0.7<br>
s2 = model.compute(C=1, V=1, T_loyalty=0.5, Z=1, sigma=0).S_official<br>
assert s2 > s1 # Z=1 должно повысить T_comp и S<br>
<br>
def test_z_influence_high_trust(model):<br>
"""Z влияет даже при высоком T_loyalty"""<br>
s1 = model.compute(C=1, V=1, T_loyalty=1, Z=0, sigma=0).S_official<br>
s2 = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0).S_official<br>
# T_comp: 0.6 vs 1.0 → значительная разница<br>
assert s2 > s1 + 10<br>
<br>
def test_extreme_low_capacity_high_vol(model):<br>
"""C=0, σ=50 → S должен быть очень низким"""<br>
result = model.compute(C=0, V=1, T_loyalty=1, Z=1, sigma=50)<br>
# S_pot≈0, F_vol≈0.167 → S≈0<br>
assert result.S_official < 1<br>
<br>
def test_extreme_high_all_zero_vol(model):<br>
"""C=V=T=Z=1, σ=0 → S≈100"""<br>
result = model.compute(C=1, V=1, T_loyalty=1, Z=1, sigma=0)<br>
assert result.S_official > 95<br>
```<br>
<br>
### Success Criteria<br>
<br>
- [ ] Написано 10 edge-case тестов<br>
- [ ] Все тесты проходят (10/10 ✅)<br>
- [ ] Все граничные случаи покрыты<br>
<br>
### Deadline<br>
<br>
**20 февраля 2026**<br>
<br>
---<br>
<br>
## 4. TASK 3: PYDANTIC ВАЛИДАЦИЯ (1 МАРТА)<br>
<br>
### Цель<br>
<br>
Защита от некорректных входов на API-уровне:<br>
- Проверка типов (float, не str)<br>
- Проверка диапазонов (C ∈ [0,1], σ ∈ [0,50])<br>
- Автоматическая ошибка 400 Bad Request при нарушении<br>
<br>
### Где применять<br>
<br>
В FastAPI endpoint (`api/app.py`).<br>
<br>
### Код<br>
<br>
```python<br>
# api/models.py<br>
from pydantic import BaseModel, Field, validator<br>
from typing import Optional<br>
<br>
class SGIndexRequest(BaseModel):<br>
"""Validated input for SG INDEX computation"""<br>
<br>
C: float = Field(<br>
...,<br>
ge=0.0,<br>
le=1.0,<br>
description="Capacity: organizational resources [0, 1]"<br>
)<br>
V: float = Field(<br>
...,<br>
ge=0.0,<br>
le=1.0,<br>
description="Visibility: social presence [0, 1]"<br>
)<br>
T_loyalty: float = Field(<br>
...,<br>
ge=0.0,<br>
le=1.0,<br>
description="Trust/Loyalty [0, 1]"<br>
)<br>
Z: float = Field(<br>
...,<br>
ge=0.0,<br>
le=1.0,<br>
description="Skepticism [0, 1]"<br>
)<br>
sigma: float = Field(<br>
...,<br>
ge=0.0,<br>
le=50.0,<br>
description="Volatility: weekly std dev [0, 50 weeks]"<br>
)<br>
<br>
@validator('*', pre=True)<br>
def check_numeric(cls, v, field):<br>
"""Ensure all fields are numeric (not str, None, etc.)"""<br>
if v is None:<br>
raise ValueError(f"{field.name} is required")<br>
if not isinstance(v, (int, float)):<br>
raise ValueError(f"{field.name} must be numeric, got {type(v).__name__}")<br>
return float(v)<br>
<br>
class Config:<br>
schema_extra = {<br>
"example": {<br>
"C": 0.8,<br>
"V": 0.7,<br>
"T_loyalty": 0.75,<br>
"Z": 0.3,<br>
"sigma": 5.0<br>
}<br>
}<br>
<br>
class SGIndexResponse(BaseModel):<br>
"""Output from SG INDEX computation"""<br>
# Inputs (echo back)<br>
C: float<br>
V: float<br>
T_loyalty: float<br>
Z: float<br>
sigma: float<br>
<br>
# Intermediate components<br>
T_comp: float<br>
S_pot: float<br>
F_gate: float<br>
F_syn: float<br>
F_vol: float<br>
S_raw: float<br>
<br>
# Final output<br>
S_official: float<br>
zone: str<br>
<br>
class Config:<br>
schema_extra = {<br>
"example": {<br>
"C": 0.8, "V": 0.7, "T_loyalty": 0.75, "Z": 0.3, "sigma": 5.0,<br>
"T_comp": 0.69, "S_pot": 0.768, "F_gate": 0.751,<br>
"F_syn": 1.178, "F_vol": 0.833, "S_raw": 0.519,<br>
"S_official": 41.2, "zone": "🟡 Caution"<br>
}<br>
}<br>
<br>
# api/app.py<br>
from fastapi import FastAPI, HTTPException<br>
from api.models import SGIndexRequest, SGIndexResponse<br>
from sg_index_v42_final import SGIndexV42<br>
<br>
app = FastAPI(title="SG INDEX v4.2 API")<br>
model = SGIndexV42()<br>
<br>
@app.post("/api/sg-index/compute", response_model=SGIndexResponse)<br>
async def compute_index(request: SGIndexRequest):<br>
"""<br>
Compute SG INDEX v4.2 for given inputs.<br>
<br>
All inputs are validated by Pydantic:<br>
- C, V, T_loyalty, Z ∈ [0, 1]<br>
- sigma ∈ [0, 50]<br>
- All must be numeric<br>
"""<br>
try:<br>
result = model.compute(<br>
C=request.C,<br>
V=request.V,<br>
T_loyalty=request.T_loyalty,<br>
Z=request.Z,<br>
sigma=request.sigma<br>
)<br>
return SGIndexResponse(**result.to_dict())<br>
except Exception as e:<br>
raise HTTPException(status_code=500, detail=str(e))<br>
<br>
@app.get("/health")<br>
async def health():<br>
"""Health check endpoint"""<br>
return {"status": "healthy", "version": "v4.2"}<br>
```<br>
<br>
### Тесты<br>
<br>
```python<br>
# tests/api/test_validation.py<br>
from fastapi.testclient import TestClient<br>
from api.app import app<br>
<br>
client = TestClient(app)<br>
<br>
def test_valid_input():<br>
"""Валидный вход → 200 OK"""<br>
response = client.post("/api/sg-index/compute", json={<br>
"C": 0.8, "V": 0.7, "T_loyalty": 0.75, "Z": 0.3, "sigma": 5.0<br>
})<br>
assert response.status_code == 200<br>
data = response.json()<br>
assert 0 <= data["S_official"] <= 100<br>
<br>
def test_invalid_c_out_of_range():<br>
"""C > 1 → 422 Validation Error"""<br>
response = client.post("/api/sg-index/compute", json={<br>
"C": 1.5, "V": 0.7, "T_loyalty": 0.75, "Z": 0.3, "sigma": 5.0<br>
})<br>
assert response.status_code == 422<br>
<br>
def test_invalid_sigma_out_of_range():<br>
"""σ > 50 → 422 Validation Error"""<br>
response = client.post("/api/sg-index/compute", json={<br>
"C": 0.8, "V": 0.7, "T_loyalty": 0.75, "Z": 0.3, "sigma": 100<br>
})<br>
assert response.status_code == 422<br>
<br>
def test_invalid_type_string():<br>
"""Строка вместо числа → 422 Validation Error"""<br>
response = client.post("/api/sg-index/compute", json={<br>
"C": "invalid", "V": 0.7, "T_loyalty": 0.75, "Z": 0.3, "sigma": 5.0<br>
})<br>
assert response.status_code == 422<br>
<br>
def test_missing_field():<br>
"""Пропущено поле → 422 Validation Error"""<br>
response = client.post("/api/sg-index/compute", json={<br>
"C": 0.8, "V": 0.7, "Z": 0.3, "sigma": 5.0<br>
# T_loyalty missing<br>
})<br>
assert response.status_code == 422<br>
```<br>
<br>
### Success Criteria<br>
<br>
- [ ] SGIndexRequest реализован с полной валидацией<br>
- [ ] API endpoint использует Pydantic models<br>
- [ ] 5 тестов валидации написаны (5/5 ✅)<br>
- [ ] API возвращает 422 на некорректные входы<br>
<br>
### Deadline<br>
<br>
**1 марта 2026**<br>
<br>
---<br>
<br>
## 5. TASK 4: CONFIG VALIDATOR (10 МАРТА)<br>
<br>
### Цель<br>
<br>
Валидация config.yaml при загрузке:<br>
- Проверка всех параметров (k, θ, ε, μ, weights)<br>
- Проверка CRS: w_C + w_T + w_V = 1.0<br>
- Защита от ручного редактирования конфига с ошибками<br>
<br>
### Код<br>
<br>
```python<br>
# utils/config_validator.py<br>
import yaml<br>
from pydantic import BaseModel, validator<br>
from typing import Dict, Any<br>
<br>
class ModelParameters(BaseModel):<br>
"""Validated model parameters"""<br>
# Theory-fixed<br>
k: float = 2.0<br>
theta: float = 0.85<br>
<br>
# Data-calibrated<br>
epsilon: float = 0.35<br>
mu: float = 0.10<br>
<br>
# Cobb-Douglas weights<br>
w_C: float = 0.25<br>
w_T: float = 0.40<br>
w_V: float = 0.35<br>
<br>
# Scaling<br>
scale_divisor: float = 1.26<br>
<br>
@validator('k')<br>
def validate_k(cls, v):<br>
if not (1.0 <= v <= 5.0):<br>
raise ValueError(f"k must be in [1.0, 5.0], got {v}")<br>
return v<br>
<br>
@validator('theta')<br>
def validate_theta(cls, v):<br>
if not (0.5 <= v <= 1.0):<br>
raise ValueError(f"theta must be in [0.5, 1.0], got {v}")<br>
return v<br>
<br>
@validator('epsilon')<br>
def validate_epsilon(cls, v):<br>
if not (0.0 <= v <= 1.0):<br>
raise ValueError(f"epsilon must be in [0.0, 1.0], got {v}")<br>
return v<br>
<br>
@validator('mu')<br>
def validate_mu(cls, v):<br>
if not (0.0 <= v <= 0.5):<br>
raise ValueError(f"mu must be in [0.0, 0.5], got {v}")<br>
return v<br>
<br>
@validator('w_C', 'w_T', 'w_V')<br>
def validate_weights(cls, v):<br>
if not (0.0 <= v <= 1.0):<br>
raise ValueError(f"weights must be in [0.0, 1.0], got {v}")<br>
return v<br>
<br>
@validator('scale_divisor')<br>
def validate_divisor(cls, v):<br>
if not (1.0 <= v <= 2.0):<br>
raise ValueError(f"scale_divisor must be in [1.0, 2.0], got {v}")<br>
return v<br>
<br>
def validate_crs(self):<br>
"""Check Constant Returns to Scale"""<br>
weight_sum = self.w_C + self.w_T + self.w_V<br>
if abs(weight_sum - 1.0) > 1e-6:<br>
raise ValueError(<br>
f"CRS violation: w_C + w_T + w_V = {weight_sum:.6f}, expected 1.0"<br>
)<br>
<br>
class SGIndexConfig(BaseModel):<br>
"""Full configuration including version, parameters, zoning"""<br>
version: str<br>
parameters: ModelParameters<br>
<br>
def validate(self):<br>
"""Run all validations"""<br>
self.parameters.validate_crs()<br>
<br>
def load_config(path: str = "config.yaml") -> SGIndexConfig:<br>
"""<br>
Load and validate configuration from YAML file.<br>
<br>
Raises:<br>
FileNotFoundError: If config file doesn't exist<br>
ValueError: If config is invalid<br>
"""<br>
with open(path, 'r') as f:<br>
data = yaml.safe_load(f)<br>
<br>
config = SGIndexConfig(<br>
version=data['version'],<br>
parameters=ModelParameters(**data['model']['parameters'])<br>
)<br>
<br>
# Validate CRS<br>
config.validate()<br>
<br>
return config<br>
<br>
# Usage example<br>
if __name__ == "__main__":<br>
try:<br>
config = load_config("config.yaml")<br>
print(f"✅ Config loaded successfully: v{config.version}")<br>
print(f" k={config.parameters.k}, θ={config.parameters.theta}")<br>
print(f" ε={config.parameters.epsilon}, μ={config.parameters.mu}")<br>
print(f" Weights: C={config.parameters.w_C}, T={config.parameters.w_T}, V={config.parameters.w_V}")<br>
print(f" CRS check: ✓ {config.parameters.w_C + config.parameters.w_T + config.parameters.w_V}")<br>
except Exception as e:<br>
print(f"❌ Config validation failed: {e}")<br>
```<br>
<br>
### Тесты<br>
<br>
```python<br>
# tests/utils/test_config_validator.py<br>
import pytest<br>
import yaml<br>
import tempfile<br>
from utils.config_validator import load_config, ModelParameters<br>
<br>
def test_valid_config():<br>
"""Валидный config → успешная загрузка"""<br>
config_data = {<br>
'version': 'v4.2',<br>
'model': {<br>
'parameters': {<br>
'k': 2.0, 'theta': 0.85, 'epsilon': 0.35, 'mu': 0.10,<br>
'w_C': 0.25, 'w_T': 0.40, 'w_V': 0.35, 'scale_divisor': 1.26<br>
}<br>
}<br>
}<br>
<br>
with tempfile.NamedTemporaryFile(mode='w', suffix='.yaml', delete=False) as f:<br>
yaml.dump(config_data, f)<br>
config_path = f.name<br>
<br>
config = load_config(config_path)<br>
assert config.version == 'v4.2'<br>
assert config.parameters.k == 2.0<br>
<br>
def test_crs_violation():<br>
"""Веса не суммируются в 1.0 → ValueError"""<br>
with pytest.raises(ValueError, match="CRS violation"):<br>
params = ModelParameters(<br>
k=2.0, theta=0.85, epsilon=0.35, mu=0.10,<br>
w_C=0.3, w_T=0.4, w_V=0.4, # сумма = 1.1 ≠ 1.0<br>
scale_divisor=1.26<br>
)<br>
params.validate_crs()<br>
<br>
def test_invalid_epsilon():<br>
"""ε > 1.0 → ValueError"""<br>
with pytest.raises(ValueError, match="epsilon"):<br>
ModelParameters(<br>
k=2.0, theta=0.85, epsilon=1.5, mu=0.10,<br>
w_C=0.25, w_T=0.40, w_V=0.35, scale_divisor=1.26<br>
)<br>
```<br>
<br>
### Success Criteria<br>
<br>
- [ ] ModelParameters реализован с полной валидацией<br>
- [ ] load_config() проверяет CRS<br>
- [ ] 3 теста написаны (3/3 ✅)<br>
- [ ] Config загружается при старте приложения с валидацией<br>
<br>
### Deadline<br>
<br>
**10 марта 2026**<br>
<br>
---<br>
<br>
## 6. TASK 5: ANTI-GAMING ML (31 МАРТА)<br>
<br>
### Цель<br>
<br>
Добавить ML-детектор аномалий для выявления манипуляций входными данными:<br>
- Использовать Isolation Forest (sklearn)<br>
- Обучить на исторических "чистых" данных<br>
- Флаг "suspicious" при отклонении от нормального паттерна<br>
<br>
### Архитектура<br>
<br>
```<br>
Входы (C, V, T, Z, σ) → Isolation Forest → Score ∈ [-1, 1]<br>
↓<br>
If score < -0.5:<br>
flag = "suspicious"<br>
```<br>
<br>
### Код<br>
<br>
```python<br>
# utils/anti_gaming.py<br>
import numpy as np<br>
from sklearn.ensemble import IsolationForest<br>
import joblib<br>
<br>
class AntiGamingDetector:<br>
"""ML-based anomaly detection for gaming attempts"""<br>
<br>
def __init__(self, model_path: str = None):<br>
if model_path:<br>
self.model = joblib.load(model_path)<br>
else:<br>
# Default untrained model<br>
self.model = IsolationForest(<br>
n_estimators=100,<br>
contamination=0.05, # 5% expected anomalies<br>
random_state=42<br>
)<br>
<br>
def fit(self, X: np.ndarray):<br>
"""<br>
Train on historical "clean" data.<br>
X shape: (n_samples, 5) for [C, V, T_loyalty, Z, sigma]<br>
"""<br>
self.model.fit(X)<br>
<br>
def predict(self, C: float, V: float, T_loyalty: float, Z: float, sigma: float) -> dict:<br>
"""<br>
Detect if inputs are suspicious.<br>
<br>
Returns:<br>
{<br>
"suspicious": bool,<br>
"score": float, # -1 (anomaly) to 1 (normal)<br>
"confidence": float # 0-1<br>
}<br>
"""<br>
X = np.array([[C, V, T_loyalty, Z, sigma]])<br>
score = self.model.score_samples(X)[0]<br>
decision = self.model.predict(X)[0] # 1 (normal), -1 (anomaly)<br>
<br>
return {<br>
"suspicious": (decision == -1),<br>
"score": float(score),<br>
"confidence": abs(score)<br>
}<br>
<br>
def save(self, path: str):<br>
"""Save trained model"""<br>
joblib.dump(self.model, path)<br>
<br>
# Training script (one-time)<br>
if __name__ == "__main__":<br>
# Load historical clean data<br>
# X_train shape: (n_samples, 5)<br>
X_train = np.array([<br>
[0.8, 0.7, 0.75, 0.3, 5.0],<br>
[0.9, 0.85, 0.8, 0.4, 3.0],<br>
# ... more samples<br>
])<br>
<br>
detector = AntiGamingDetector()<br>
detector.fit(X_train)<br>
detector.save("models/anti_gaming_v1.pkl")<br>
print("✅ Model trained and saved")<br>
<br>
# Integration into API<br>
from utils.anti_gaming import AntiGamingDetector<br>
<br>
# In api/app.py<br>
detector = AntiGamingDetector(model_path="models/anti_gaming_v1.pkl")<br>
<br>
@app.post("/api/sg-index/compute", response_model=SGIndexResponse)<br>
async def compute_index(request: SGIndexRequest):<br>
# Check for gaming<br>
gaming_check = detector.predict(<br>
C=request.C, V=request.V, T_loyalty=request.T_loyalty,<br>
Z=request.Z, sigma=request.sigma<br>
)<br>
<br>
if gaming_check["suspicious"]:<br>
# Log for review, but still compute<br>
logger.warning(f"Suspicious input detected: {request.dict()}, score={gaming_check['score']}")<br>
<br>
result = model.compute(...)<br>
response = SGIndexResponse(**result.to_dict())<br>
<br>
# Add gaming flag to response<br>
response.gaming_flag = gaming_check["suspicious"]<br>
response.gaming_score = gaming_check["score"]<br>
<br>
return response<br>
```<br>
<br>
### Success Criteria<br>
<br>
- [ ] AntiGamingDetector реализован<br>
- [ ] Модель обучена на ~1000 historical samples<br>
- [ ] Интегрирована в API (флаг "suspicious" в response)<br>
- [ ] Логирование подозрительных запросов<br>
<br>
### Deadline<br>
<br>
**31 марта 2026**<br>
<br>
---<br>
<br>
## 7. TIMELINE (GANTT)<br>
<br>
```<br>
Февраль 2026:<br>
┌──────────────────────────────────────────────────┐<br>
│ Week 1 (3-9 Feb): Task 1 (unit-тесты) │ ████████░░░░░░<br>
│ Week 2 (10-16 Feb): Task 1 завершение │ ░░░░░░░░████░░<br>
│ Week 3 (17-23 Feb): Task 2 (edge-cases) │ ░░░░░░░░░░░░████<br>
│ Week 4 (24-28 Feb): Weekly pilot report #1 │<br>
└──────────────────────────────────────────────────┘<br>
<br>
Март 2026:<br>
┌──────────────────────────────────────────────────┐<br>
│ Week 1 (3-9 Mar): Task 3 (Pydantic) │ ████████░░░░░░<br>
│ Week 2 (10-16 Mar): Task 4 (config validator) │ ░░░░░░░░████░░<br>
│ Week 3 (17-23 Mar): Task 5 (anti-gaming) part 1 │ ░░░░░░░░░░░░████<br>
│ Week 4 (24-31 Mar): Task 5 завершение + v4.3 │ ░░░░░░░░░░░░░░░░████<br>
└──────────────────────────────────────────────────┘<br>
<br>
Milestones:<br>
├─ 15 Feb: Unit-тесты done (20/20 ✅)<br>
├─ 20 Feb: Edge-cases done (10/10 ✅)<br>
├─ 01 Mar: Pydantic done<br>
├─ 10 Mar: Config validator done<br>
└─ 31 Mar: P1 Release (v4.3) + Weekly pilot report #2<br>
```<br>
<br>
---<br>
<br>
## 8. SUCCESS CRITERIA (KPI)<br>
<br>
| Task | Метрика | Target | Measurement |<br>
|------|---------|--------|-------------|<br>
| Task 1 | Coverage | ≥ 85% | `pytest --cov` |<br>
| Task 1 | Unit-тесты | 20/20 ✅ | `pytest tests/unit/` |<br>
| Task 2 | Edge-case тесты | 10/10 ✅ | `pytest tests/unit/test_edge_cases.py` |<br>
| Task 3 | API validation | 5/5 ✅ | `pytest tests/api/test_validation.py` |<br>
| Task 4 | Config validation | 3/3 ✅ | `pytest tests/utils/test_config_validator.py` |<br>
| Task 5 | Anti-gaming | Model trained | Accuracy ≥ 95% на test set |<br>
| **OVERALL** | **P1 Completion** | **100%** | **All 5 tasks done by 31 Mar** |<br>
<br>
---<br>
<br>
## 9. ЗАКЛЮЧЕНИЕ<br>
<br>
**P1 Roadmap готов к выполнению.**<br>
<br>
После завершения всех 5 задач:<br>
- Coverage вырастет с 75% → 85%+<br>
- Код получит оценку 10/10 (вместо 9/10)<br>
- Тестирование получит оценку 10/10 (вместо 8/10)<br>
- **Общая оценка:** 9.4 → **9.8/10** ✅<br>
<br>
**Это повысит долгосрочную стабильность модели и упростит дальнейшее развитие (P2).**<br>
<br>
---<br>
<br>
**Подготовлено:** Development Team <br>
**Дата:** 10 января 2026 <br>
**Статус:** 📅 **PLANNED** (Feb-Mar 2026) <br>
**Приоритет:** P1 (High, но не блокирует launch)<br>