QuantEngineByItz/tools/evaluate_qualitative_sell_strategy_accuracy_v1.py

"""qualitative_sell_strategy_v1 자체 평가 루프 — "한 번 만들고 끝"이 아니라 결정이
실제로 가치를 보존했는지 사후 검증한다(30년 시니어 퀀트의 핵심 습관: 판단 → 결과 →
재보정). 기존 T+5/T+20 outcome ledger(proposal_evaluation_history)와 별개로,
qualitative_sell_strategy_store_v1.db에 쌓인 SQLite 시계열을 사용한다 — GAS/xlsx와
무관하므로 이 모듈만의 독립 평가 루프를 구성해도 기존 시스템과 충돌하지 않는다.

판정 기준(가치보존 관점, 기계적 승률 게임이 아님):
  - EXIT_REVIEW_FULL / TRIM_REVIEW_PARTIAL(매도방향) → 이후 가격이 하락했으면
    "가치보존 성공"(매도가 손실을 막았다). 상승했으면 "기회비용 발생"(조급한 매도).
  - HOLD_ADD_CONVICTION(지지방향) → 이후 가격이 상승했으면 성공.
  - HOLD_NO_CONFLUENCE / INSUFFICIENT_DATA_NO_ACTION → 방향성 주장이 없으므로 평가 대상 제외.

표본이 부족하면(DATA_GATED) 추정하지 않고 명시적으로 보류한다 — honest_proof_score와
동일한 원칙(spec/algorithm_guidance_proof 계열).
"""
from __future__ import annotations

import argparse
import datetime as dt
import json
import sqlite3
import sys
from pathlib import Path
from typing import Any

ROOT = Path(__file__).resolve().parents[1]
if str(ROOT) not in sys.path:
    sys.path.insert(0, str(ROOT))

from src.quant_engine.qualitative_sell_strategy_store_v1 import QualitativeSellStoreSpec, resolve_store_path

MIN_HOLDING_DAYS = 5  # T+5 수준 — 너무 짧으면 노이즈, 너무 길면 표본 희소
MIN_SAMPLE_FOR_HIT_RATE = 10  # 이보다 적으면 hit_rate를 신뢰 구간 없이 표기하지 않음(DATA_GATED)


def _scoreable_direction(action: str) -> int | None:
    if action in {"EXIT_REVIEW_FULL", "TRIM_REVIEW_PARTIAL"}:
        return -1  # 매도 방향 — 가격 하락이 "성공"
    if action == "HOLD_ADD_CONVICTION":
        return 1  # 지지 방향 — 가격 상승이 "성공"
    return None  # HOLD_NO_CONFLUENCE / INSUFFICIENT_DATA_NO_ACTION — 평가 제외


def load_scoreable_decisions(db_path: Path, min_age_days: int = MIN_HOLDING_DAYS) -> list[dict[str, Any]]:
    if not db_path.exists():
        return []
    cutoff = (dt.date.today() - dt.timedelta(days=min_age_days)).isoformat()
    conn = sqlite3.connect(db_path)
    conn.row_factory = sqlite3.Row
    try:
        rows = conn.execute(
            "SELECT code, generated_at, action, conviction, market_regime, composite_score "
            "FROM sell_strategy_results WHERE generated_at <= ? ORDER BY generated_at",
            (cutoff,),
        ).fetchall()
        return [dict(row) for row in rows]
    finally:
        conn.close()


def evaluate_decision(decision: dict[str, Any], price_at_decision: float, price_after: float) -> dict[str, Any] | None:
    direction = _scoreable_direction(decision["action"])
    if direction is None or not price_at_decision or price_at_decision <= 0:
        return None
    realized_return_pct = (price_after / price_at_decision - 1.0) * 100.0
    success = (direction * realized_return_pct) > 0  # 방향 일치 시 성공
    return {
        **decision,
        "price_at_decision": price_at_decision,
        "price_after": price_after,
        "realized_return_pct": round(realized_return_pct, 4),
        "success": success,
    }


def build_accuracy_report(db_path: Path, price_lookup: dict[str, dict[str, float]]) -> dict[str, Any]:
    """price_lookup: {code: {generated_at_date_iso: close_price}} — 호출측이 실제 가격
    히스토리(fetch_naver_market_data_v1 등)로 조립해 주입한다. 이 함수는 가격을 추정하지
    않는다 — 주어진 값만 사용."""
    decisions = load_scoreable_decisions(db_path)
    evaluated: list[dict[str, Any]] = []
    skipped_no_price = 0
    for decision in decisions:
        prices = price_lookup.get(decision["code"], {})
        decision_date = decision["generated_at"][:10]
        price_at = prices.get(decision_date)
        future_date = (dt.date.fromisoformat(decision_date) + dt.timedelta(days=MIN_HOLDING_DAYS)).isoformat()
        price_after = prices.get(future_date)
        if price_at is None or price_after is None:
            skipped_no_price += 1
            continue
        result = evaluate_decision(decision, price_at, price_after)
        if result is not None:
            evaluated.append(result)

    scored = [e for e in evaluated if e is not None]
    if len(scored) < MIN_SAMPLE_FOR_HIT_RATE:
        return {
            "status": "DATA_GATED",
            "scored_sample_count": len(scored),
            "min_sample_required": MIN_SAMPLE_FOR_HIT_RATE,
            "note": "표본 부족 — hit_rate를 산출하지 않음(추정 금지). 결정 누적과 가격 매칭이 더 필요.",
            "skipped_no_price": skipped_no_price,
        }

    hit_rate_pct = round(100.0 * sum(1 for e in scored if e["success"]) / len(scored), 2)
    return {
        "status": "OK",
        "scored_sample_count": len(scored),
        "hit_rate_pct": hit_rate_pct,
        "evaluations": scored,
        "skipped_no_price": skipped_no_price,
    }


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--sqlite-db", type=Path,
                     default=ROOT / "outputs" / "qualitative_sell_strategy" / "qualitative_sell_strategy.db")
    ap.add_argument("--store-backend", default="sqlite", help="Storage backend contract placeholder (sqlite today, postgresql planned)")
    ap.add_argument("--store-location", default=None, help="Backend location/DSN. sqlite path or future postgres DSN.")
    ap.add_argument("--price-lookup-json", type=Path, default=None,
                     help='{"code": {"YYYY-MM-DD": close_price, ...}} 형식 — 미지정 시 가격 매칭 없이 표본 카운트만 보고')
    args = ap.parse_args()
    db_path = resolve_store_path(
        QualitativeSellStoreSpec(
            backend=args.store_backend,
            location=args.store_location or args.sqlite_db,
        ),
        ROOT,
    )

    price_lookup: dict[str, dict[str, float]] = {}
    if args.price_lookup_json and args.price_lookup_json.exists():
        price_lookup = json.loads(args.price_lookup_json.read_text(encoding="utf-8"))

    report = build_accuracy_report(db_path, price_lookup)
    print(json.dumps(report, ensure_ascii=False, indent=2))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())