paste.txt

ChatGPT neutral 11 чанков ~15 мин чтения
# 📝 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>