htlee
8f5152fc02
런타임 override 완성 (params 인자 + 내부 상수 교체): - gear_violation_g01_g06 (GEAR, tier 4) · G01~G06 점수 + signal_cycling(gap_min/min_count) · gear_drift_threshold_nm + fixed_gear_types + fishery_code_allowed_gear · _detect_signal_cycling_count 도입 (기존 _detect_signal_cycling 보존) 카탈로그 + 관찰 (DEFAULT_PARAMS 노출 + Adapter 집계, 런타임 교체는 후속 PR): - transshipment_5stage (TRANSSHIP, tier 4) — 5단계 필터 임계 - risk_composite (META, tier 3) — 경량+파이프라인 가중치 - pair_trawl_tier (GEAR, tier 4) — STRONG/PROBABLE/SUSPECT 임계 각 모델 공통: - prediction/algorithms/*.py: DEFAULT_PARAMS 상수 추가 - models_core/registered/*_model.py: BaseDetectionModel Adapter - models_core/seeds/v1_<model>.sql: DRAFT seed (호출자 트랜잭션 제어) - tests/test_<model>_params.py: Python ↔ 모듈 상수 ↔ seed SQL 정적 일치 검증 통합 seed: models_core/seeds/v1_phase2_all.sql (\i 로 5 모델 일괄 시드) 검증: - 30/30 테스트 통과 (Phase 1-2 15 + dark 5 + Phase 2 신규 10) - 운영 DB 5 모델 개별 + 일괄 seed dry-run 통과 (BEGIN/ROLLBACK 격리) - 5 모델 모두 tier/category 정렬 확인: dark_suspicion(3) / risk_composite(3) / gear_violation_g01_g06(4) / pair_trawl_tier(4) / transshipment_5stage(4) 후속: - transshipment/risk/pair_trawl 런타임 override 활성화 (헬퍼 params 전파) - Phase 3 백엔드 API (DetectionModelController + 승격 엔드포인트) Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
76 lines
2.6 KiB
Python
76 lines
2.6 KiB
Python
"""Phase 2 PoC #4 — risk_composite DEFAULT_PARAMS ↔ seed SQL 정적 일치 테스트."""
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import types
|
|
import unittest
|
|
|
|
if 'pandas' not in sys.modules:
|
|
pd_stub = types.ModuleType('pandas')
|
|
pd_stub.DataFrame = type('DataFrame', (), {})
|
|
pd_stub.Timestamp = type('Timestamp', (), {})
|
|
sys.modules['pandas'] = pd_stub
|
|
|
|
if 'pydantic_settings' not in sys.modules:
|
|
stub = types.ModuleType('pydantic_settings')
|
|
|
|
class _S:
|
|
def __init__(self, **kw):
|
|
for name, value in self.__class__.__dict__.items():
|
|
if name.isupper():
|
|
setattr(self, name, kw.get(name, value))
|
|
|
|
stub.BaseSettings = _S
|
|
sys.modules['pydantic_settings'] = stub
|
|
|
|
if 'algorithms' not in sys.modules:
|
|
pkg = types.ModuleType('algorithms')
|
|
pkg.__path__ = [os.path.join(os.path.dirname(__file__), '..', 'algorithms')]
|
|
sys.modules['algorithms'] = pkg
|
|
|
|
# risk.py 는 algorithms.location/fishing_pattern/dark_vessel/spoofing 을 top-level
|
|
# import 한다. dark_vessel 만 실제 모듈 그대로 두고 나머지는 필요한 심볼만 stub.
|
|
if 'algorithms.location' in sys.modules:
|
|
if not hasattr(sys.modules['algorithms.location'], 'classify_zone'):
|
|
sys.modules['algorithms.location'].classify_zone = lambda *a, **k: {}
|
|
else:
|
|
loc = types.ModuleType('algorithms.location')
|
|
loc.haversine_nm = lambda a, b, c, d: 0.0
|
|
loc.classify_zone = lambda *a, **k: {}
|
|
sys.modules['algorithms.location'] = loc
|
|
|
|
for mod_name, attrs in [
|
|
('algorithms.fishing_pattern', ['detect_fishing_segments', 'detect_trawl_uturn']),
|
|
('algorithms.spoofing', ['detect_teleportation', 'count_speed_jumps']),
|
|
]:
|
|
if mod_name not in sys.modules:
|
|
m = types.ModuleType(mod_name)
|
|
sys.modules[mod_name] = m
|
|
m = sys.modules[mod_name]
|
|
for a in attrs:
|
|
if not hasattr(m, a):
|
|
setattr(m, a, lambda *_a, **_kw: [])
|
|
|
|
|
|
class RiskCompositeParamsTest(unittest.TestCase):
|
|
|
|
def test_seed_matches_default(self):
|
|
risk = importlib.import_module('algorithms.risk')
|
|
seed_path = os.path.join(
|
|
os.path.dirname(__file__), '..',
|
|
'models_core', 'seeds', 'v1_risk_composite.sql',
|
|
)
|
|
with open(seed_path, 'r', encoding='utf-8') as f:
|
|
sql = f.read()
|
|
start = sql.index('$json$') + len('$json$')
|
|
end = sql.index('$json$', start)
|
|
params = json.loads(sql[start:end].strip())
|
|
self.assertEqual(params, risk.RISK_COMPOSITE_DEFAULT_PARAMS)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|