Bithumb/deepcoin/matching/gt_asset_calibration.py

"""
GT 총자산 대비 시뮬/규칙 정확도 측정 (동일 체결·평가 모델).
"""

from __future__ import annotations

from typing import Any

import pandas as pd

from config import GT_INITIAL_CASH_KRW, MATCH_GT_TOLERANCE_MIN, TRADING_FEE_RATE
from deepcoin.ground_truth.ground_truth import simulate_truth_portfolio
from deepcoin.matching.rule_eval import eval_rule_mask


def gt_trades_for_legs(
    trades: list[dict[str, Any]],
    leg_ids: set[int],
) -> list[dict[str, Any]]:
    """
    leg_id 집합에 속한 GT 체결만 반환.

    Args:
        trades: ground_truth trades.
        leg_ids: 포함할 leg_id.

    Returns:
        필터된 trade dict 리스트.
    """
    return [t for t in trades if int(t.get("leg_id", 0)) in leg_ids]


def covered_legs_from_fires(
    trades: list[dict[str, Any]],
    fires: pd.DataFrame,
    buy_rule_ids: list[str],
    sell_rule_ids: list[str],
    tolerance_min: int = MATCH_GT_TOLERANCE_MIN,
) -> set[int]:
    """
    매수·매도 규칙 발화가 GT 타점 ±허용 내인 leg_id 집합.

    Args:
        trades: GT trades.
        fires: rule_fires.
        buy_rule_ids: 매수 규칙 ID.
        sell_rule_ids: 매도 규칙 ID.
        tolerance_min: 허용 분.

    Returns:
        양쪽 모두 커버된 leg_id.
    """
    if fires.empty:
        return set()
    tol = pd.Timedelta(minutes=tolerance_min)
    gt_df = pd.DataFrame(trades)
    gt_df["ts"] = pd.to_datetime(gt_df["dt"])
    fires = fires.copy()
    fires["ts"] = pd.to_datetime(fires["dt"])
    bf = fires[fires["rule_id"].isin(buy_rule_ids) & (fires["side"] == "buy")]
    sf = fires[fires["rule_id"].isin(sell_rule_ids) & (fires["side"] == "sell")]

    covered: set[int] = set()
    for lid in gt_df["leg_id"].unique():
        leg = gt_df[gt_df["leg_id"] == lid]
        buys = leg[leg["action"] == "buy"]
        sells = leg[leg["action"] == "sell"]
        buy_ok = True
        for ts in buys["ts"]:
            if bf.empty or (bf["ts"] - ts).abs().min() > tol:
                buy_ok = False
                break
        sell_ok = True
        for ts in sells["ts"]:
            if sf.empty or (sf["ts"] - ts).abs().min() > tol:
                sell_ok = False
                break
        if buy_ok and sell_ok:
            covered.add(int(lid))
    return covered


def portfolio_asset_ratio(
    trades: list[dict[str, Any]],
    leg_ids: set[int],
    last_price: float | None,
) -> dict[str, Any]:
    """
    GT 체결 모델로 전체 vs 부분 leg 포트폴리오 비율.

    Args:
        trades: 전체 GT trades.
        leg_ids: 포함 leg.
        last_price: 종가 평가.

    Returns:
        full/subset final_asset, asset_ratio, leg counts.
    """
    full = simulate_truth_portfolio(
        trades,
        initial_cash=GT_INITIAL_CASH_KRW,
        fee_rate=TRADING_FEE_RATE,
        last_price=last_price,
    )
    subset_trades = gt_trades_for_legs(trades, leg_ids)
    part = simulate_truth_portfolio(
        subset_trades,
        initial_cash=GT_INITIAL_CASH_KRW,
        fee_rate=TRADING_FEE_RATE,
        last_price=last_price,
    )
    gt_final = float(full["final_asset_krw"])
    sub_final = float(part["final_asset_krw"])
    ratio = sub_final / gt_final if gt_final > 0 else 0.0
    return {
        "gt_final_asset_krw": gt_final,
        "subset_final_asset_krw": sub_final,
        "asset_ratio": round(ratio, 4),
        "asset_accuracy_pct": round(ratio * 100.0, 2),
        "target_met_90": ratio >= 0.9,
        "legs_total": len(set(int(t.get("leg_id", 0)) for t in trades)),
        "legs_covered": len(leg_ids),
        "leg_coverage_ratio": round(
            len(leg_ids) / max(len(set(int(t.get("leg_id", 0)) for t in trades)), 1),
            4,
        ),
        "full_pnl_pct": full.get("pnl_pct"),
        "subset_pnl_pct": part.get("pnl_pct"),
    }


def evaluate_gt_snapshot_recall(
    trades_df: pd.DataFrame,
    rules: list[dict[str, Any]],
) -> dict[str, Any]:
    """
    03b 각 GT 행에서 규칙 스냅샷 충족 여부(OR across rules per side).

    Args:
        trades_df: general_analysis_trades.csv.
        rules: rule dict 리스트.

    Returns:
        buy/sell recall, per-rule counts.
    """
    buy_gt = trades_df[trades_df["action"] == "buy"]
    sell_gt = trades_df[trades_df["action"] == "sell"]
    buy_rules = [r for r in rules if r.get("side") == "buy"]
    sell_rules = [r for r in rules if r.get("side") == "sell"]

    def _side_recall(gt: pd.DataFrame, side_rules: list[dict]) -> dict[str, Any]:
        if gt.empty or not side_rules:
            return {"gt_count": int(len(gt)), "matched": 0, "recall": 0.0}
        hit = 0
        per_rule: dict[str, int] = {}
        for _, row in gt.iterrows():
            fr = pd.DataFrame([row])
            ok = False
            for rule in side_rules:
                if bool(eval_rule_mask(fr, rule).iloc[0]):
                    ok = True
                    rid = rule["rule_id"]
                    per_rule[rid] = per_rule.get(rid, 0) + 1
            if ok:
                hit += 1
        n = len(gt)
        return {
            "gt_count": n,
            "matched": hit,
            "recall": round(hit / n, 4) if n else 0.0,
            "per_rule_hits": per_rule,
        }

    return {
        "buy": _side_recall(buy_gt, buy_rules),
        "sell": _side_recall(sell_gt, sell_rules),
    }