kcg-ai-monitoring/prediction/algorithms/risk.py

from typing import Optional, Tuple

import pandas as pd
from algorithms.location import classify_zone
from algorithms.fishing_pattern import detect_fishing_segments, detect_trawl_uturn
from algorithms.dark_vessel import detect_ais_gaps
from algorithms.spoofing import detect_teleportation


# Phase 2 PoC #4 — risk_composite 카탈로그 등록용 params snapshot.
# 현 런타임은 모듈 레벨 상수/inline 숫자를 직접 사용하며, 운영자 UI 에서
# 주요 가중치·임계를 조회·튜닝할 수 있도록 DB 에 노출한다. 런타임 override
# 는 후속 리팩토링 PR 에서 compute_lightweight_risk_score / compute_vessel_risk_score
# 에 params 인자 전파를 완성하면서 활성화된다.
RISK_COMPOSITE_DEFAULT_PARAMS: dict = {
    'tier_thresholds': {'critical': 70, 'high': 50, 'medium': 30},
    # 경량(파이프라인 미통과) 경로 — compute_lightweight_risk_score
    'lightweight_weights': {
        'territorial_sea': 40,
        'contiguous_zone': 15,
        'zone_unpermitted': 25,
        'eez_lt12nm': 15,
        'eez_lt24nm': 8,
        'dark_suspicion_multiplier': 0.3,
        'dark_gap_720_min': 25,
        'dark_gap_180_min': 20,
        'dark_gap_60_min': 15,
        'dark_gap_30_min': 8,
        'spoofing_gt07': 15,
        'spoofing_gt05': 8,
        'unpermitted_alone': 15,
        'unpermitted_with_suspicion': 8,
        'repeat_gte5': 10,
        'repeat_gte2': 5,
    },
    # 파이프라인 통과(정밀) 경로 — compute_vessel_risk_score
    'pipeline_weights': {
        'territorial_sea': 40,
        'contiguous_zone': 10,
        'zone_unpermitted': 25,
        'territorial_fishing': 20,
        'fishing_segments_any': 5,
        'trawl_uturn': 10,
        'teleportation': 20,
        'speed_jumps_ge3': 10,
        'speed_jumps_ge1': 5,
        'critical_gaps_ge60': 15,
        'any_gaps': 5,
        'unpermitted': 20,
    },
    'dark_suspicion_fallback_gap_min': {
        'very_long_720': 720,
        'long_180': 180,
        'mid_60': 60,
        'short_30': 30,
    },
    'spoofing_thresholds': {'high_0.7': 0.7, 'medium_0.5': 0.5},
    'eez_proximity_nm': {'inner_12': 12, 'outer_24': 24},
    'repeat_thresholds': {'h24_high': 5, 'h24_low': 2},
}


def compute_lightweight_risk_score(
    zone_info: dict,
    sog: float,
    is_permitted: Optional[bool] = None,
    is_dark: bool = False,
    gap_duration_min: int = 0,
    spoofing_score: float = 0.0,
    dark_suspicion_score: int = 0,
    dist_from_baseline_nm: float = 999.0,
    dark_history_24h: int = 0,
) -> Tuple[int, str]:
    """위치·허가·다크/스푸핑 기반 경량 위험도 (파이프라인 미통과 선박용).

    compute_dark_suspicion 의 패턴 기반 0~100 점수를 직접 반영해 해상도를 높인다.
    이중계산 방지: dark_suspicion_score 는 이미 무허가/반복을 포함하므로 dark_suspicion_score > 0
    인 경우 허가/반복 가산을 축소한다.

    임계값 70/50/30 은 pipeline path(compute_vessel_risk_score)와 동일.

    Returns: (risk_score, risk_level)
    """
    score = 0

    # 1. 위치 기반 (최대 40점) — EEZ 외 기선 근접도 추가
    zone = zone_info.get('zone', '')
    if zone == 'TERRITORIAL_SEA':
        score += 40
    elif zone == 'CONTIGUOUS_ZONE':
        score += 15
    elif zone.startswith('ZONE_'):
        if is_permitted is not None and not is_permitted:
            score += 25
    elif zone == 'EEZ_OR_BEYOND':
        # EEZ 외라도 기선 근접 시 가산 (공해·외해 분산)
        if dist_from_baseline_nm < 12:
            score += 15
        elif dist_from_baseline_nm < 24:
            score += 8

    # 2. 다크 베셀 (최대 30점) — dark_suspicion_score 우선
    if is_dark:
        if dark_suspicion_score >= 1:
            # compute_dark_suspicion 이 산출한 패턴 기반 의심도 반영
            score += min(30, round(dark_suspicion_score * 0.3))
        else:
            # fallback: gap 길이만 기준
            if gap_duration_min >= 720:
                score += 25
            elif gap_duration_min >= 180:
                score += 20
            elif gap_duration_min >= 60:
                score += 15
            elif gap_duration_min >= 30:
                score += 8

    # 3. 스푸핑 (최대 15점) — 현재 중국 선박은 거의 0 (별도 PR 에서 산출 로직 재설계 예정)
    if spoofing_score > 0.7:
        score += 15
    elif spoofing_score > 0.5:
        score += 8

    # 4. 허가 이력 (최대 15점) — 이중계산 방지
    if is_permitted is not None and not is_permitted:
        # dark_suspicion_score 에 이미 무허가 +10 반영됨 → 축소
        score += 8 if dark_suspicion_score > 0 else 15

    # 5. 반복 이력 (최대 10점) — dark_suspicion_score 미반영 케이스만
    if dark_suspicion_score == 0 and dark_history_24h > 0:
        if dark_history_24h >= 5:
            score += 10
        elif dark_history_24h >= 2:
            score += 5

    score = min(score, 100)

    if score >= 70:
        level = 'CRITICAL'
    elif score >= 50:
        level = 'HIGH'
    elif score >= 30:
        level = 'MEDIUM'
    else:
        level = 'LOW'

    return score, level


def compute_vessel_risk_score(
    mmsi: str,
    df_vessel: pd.DataFrame,
    zone_info: Optional[dict] = None,
    is_permitted: Optional[bool] = None,
) -> Tuple[int, str]:
    """선박별 종합 위반 위험도 (0~100점).

    Returns: (risk_score, risk_level)
    """
    if len(df_vessel) == 0:
        return 0, 'LOW'

    score = 0

    # 1. 위치 기반 (최대 40점)
    if zone_info is None:
        last = df_vessel.iloc[-1]
        zone_info = classify_zone(last['lat'], last['lon'])

    zone = zone_info.get('zone', '')
    if zone == 'TERRITORIAL_SEA':
        score += 40
    elif zone == 'CONTIGUOUS_ZONE':
        score += 10
    elif zone.startswith('ZONE_'):
        # 특정어업수역 내 — 무허가면 가산
        if is_permitted is not None and not is_permitted:
            score += 25

    # 2. 조업 행위 (최대 30점)
    segs = detect_fishing_segments(df_vessel)
    ts_fishing = [s for s in segs if s.get('in_territorial_sea')]
    if ts_fishing:
        score += 20
    elif segs:
        score += 5

    uturn = detect_trawl_uturn(df_vessel)
    if uturn.get('trawl_suspected'):
        score += 10

    # 3. AIS 조작 (최대 35점)
    teleports = detect_teleportation(df_vessel)
    if teleports:
        score += 20

    from algorithms.spoofing import count_speed_jumps
    jumps = count_speed_jumps(df_vessel)
    if jumps >= 3:
        score += 10
    elif jumps >= 1:
        score += 5

    gaps = detect_ais_gaps(df_vessel)
    critical_gaps = [g for g in gaps if g['gap_min'] >= 60]
    if critical_gaps:
        score += 15
    elif gaps:
        score += 5

    # 4. 허가 이력 (최대 20점)
    if is_permitted is not None and not is_permitted:
        score += 20

    score = min(score, 100)

    if score >= 70:
        level = 'CRITICAL'
    elif score >= 50:
        level = 'HIGH'
    elif score >= 30:
        level = 'MEDIUM'
    else:
        level = 'LOW'

    return score, level