QuantEngineByItz/tools/build_sell_waterfall_engine_v4.py

#!/usr/bin/env python3
"""SELL_LOT_PARETO_SELECTOR_V1 — spec/formulas/domains/cash.yaml.

Extends tools/build_sell_waterfall_engine_v3.py output with lot-level scoring
(tax_loss_benefit, missed_upside_penalty, reentry_cost) and a Pareto dominance
ranking within each hard_precedence stage, per
governance/todo/v8_9_p0_adoption_plan.yaml P0-2.2.

Backward compatible: every row from v3 is preserved unchanged; only new fields
are appended.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_V3 = ROOT / "Temp" / "sell_waterfall_engine_v3.json"
DEFAULT_OUT = ROOT / "Temp" / "sell_waterfall_engine_v4.json"

MAXIMIZE_FIELDS = [
    "avoided_tail_loss_krw",
    "cash_repair_benefit_krw",
    "concentration_reduction_benefit_krw",
    "tax_loss_benefit_krw",
]
MINIMIZE_FIELDS = [
    "tax_fee_slippage_krw",
    "reentry_cost_krw",
    "missed_upside_penalty_krw",
]


def _load(path: Path) -> dict:
    if not path.exists():
        return {}
    try:
        data = json.loads(path.read_text(encoding="utf-8"))
        return data if isinstance(data, dict) else {}
    except Exception:
        return {}


def _numeric_or_missing(row: dict, field: str) -> tuple[float, bool]:
    value = row.get(field)
    if value is None:
        return 0.0, True
    try:
        return float(value), False
    except (TypeError, ValueError):
        return 0.0, True


def _lot_sell_score(row: dict) -> tuple[float, list[str]]:
    missing_fields = []
    values = {}
    for field in MAXIMIZE_FIELDS + MINIMIZE_FIELDS:
        value, missing = _numeric_or_missing(row, field)
        values[field] = value
        if missing:
            missing_fields.append(field)
    score = (
        values["avoided_tail_loss_krw"]
        + values["cash_repair_benefit_krw"]
        + values["concentration_reduction_benefit_krw"]
        + values["tax_loss_benefit_krw"]
        - values["tax_fee_slippage_krw"]
        - values["reentry_cost_krw"]
        - values["missed_upside_penalty_krw"]
    )
    return score, missing_fields


def _dominates(a: dict, b: dict) -> bool:
    at_least_as_good = all(a.get(f, 0.0) >= b.get(f, 0.0) for f in MAXIMIZE_FIELDS) and all(
        a.get(f, 0.0) <= b.get(f, 0.0) for f in MINIMIZE_FIELDS
    )
    strictly_better = any(a.get(f, 0.0) > b.get(f, 0.0) for f in MAXIMIZE_FIELDS) or any(
        a.get(f, 0.0) < b.get(f, 0.0) for f in MINIMIZE_FIELDS
    )
    return at_least_as_good and strictly_better


def _rank_pareto_group(rows: list[dict]) -> list[dict]:
    annotated = []
    for row in rows:
        dominated_by = [
            other["candidate_id"]
            for other in rows
            if other is not row and _dominates(other, row)
        ]
        annotated.append({**row, "pareto_dominated": bool(dominated_by), "dominated_by": dominated_by})
    annotated.sort(
        key=lambda r: (
            r["pareto_dominated"],
            -r["lot_sell_score_krw"],
            r.get("tax_fee_slippage_krw", 0.0),
            r.get("reentry_cost_krw", 0.0),
        )
    )
    for idx, row in enumerate(annotated, start=1):
        row["pareto_rank"] = idx
    return annotated


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default=str(DEFAULT_V3))
    ap.add_argument("--out", default=str(DEFAULT_OUT))
    args = ap.parse_args()

    base = _load(Path(args.base))
    rows = base.get("rows") if isinstance(base.get("rows"), list) else []

    scored_rows = []
    for idx, row in enumerate(rows):
        if not isinstance(row, dict):
            continue
        candidate_id = row.get("candidate_id") or row.get("종목명") or f"LOT_{idx}"
        score, missing_fields = _lot_sell_score(row)
        scored_rows.append(
            {
                **row,
                "candidate_id": candidate_id,
                **{f: _numeric_or_missing(row, f)[0] for f in MAXIMIZE_FIELDS + MINIMIZE_FIELDS},
                "lot_sell_score_krw": score,
                "lot_sell_score_missing_fields": missing_fields,
                "hard_precedence_stage": row.get("hard_precedence_stage") or row.get("우선순위단계"),
            }
        )

    groups: dict[object, list[dict]] = {}
    for row in scored_rows:
        groups.setdefault(row.get("hard_precedence_stage"), []).append(row)

    ranked_rows: list[dict] = []
    for _stage, group_rows in groups.items():
        ranked_rows.extend(_rank_pareto_group(group_rows))

    result = {
        "formula_id": "SELL_LOT_PARETO_SELECTOR_V1",
        "gate": base.get("gate") or "PASS",
        "rows": ranked_rows,
        "source_paths": [str(Path(args.base))],
    }
    out = Path(args.out)
    out.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
    print(json.dumps(result, ensure_ascii=False, indent=2))
    return 0


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