QuantEngineByItz/tools/build_sector_exposure_graph_v1.py

#!/usr/bin/env python3
"""SECTOR_EXPOSURE_GRAPH_V1 + LEADER_LIFECYCLE_GATE_V1 — spec/formulas/domains/sector.yaml.

ETF lookthrough exposure + factor beta residualization + leader role promotion/demotion.
governance/todo/v8_9_p1_adoption_plan.yaml P1-A.3.

Hard rule (AGENTS.md): missing ETF constituents or peer betas are never assumed zero.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_POSITIONS = ROOT / "Temp" / "sector_exposure_positions_v1.json"
DEFAULT_OUT = ROOT / "Temp" / "sector_exposure_graph_v1.json"

PROMOTION_PATH = ["LAGGARD", "CYCLICAL_BETA", "ENABLER", "CORE_LEADER", "CAPTAIN"]


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 sector_exposure(position: dict) -> dict:
    direct_weight_pct = float(position.get("direct_weight_pct") or 0.0)
    etf_constituents = position.get("etf_constituents_json")
    etf_weight_pct = position.get("etf_weight_pct")
    sector_id = position.get("sector_id")

    if etf_constituents is None or etf_weight_pct is None:
        return {
            "sector_id": sector_id,
            "direct_weight_pct": direct_weight_pct,
            "lookthrough_etf_weight_pct": None,
            "sector_family_total_pct": None,
            "gate": "ETF_BUY_BLOCKED",
            "reason_code": "constituents_missing",
        }

    lookthrough = sum(
        float(c.get("weight_pct", 0.0)) * float(etf_weight_pct) / 100.0
        for c in etf_constituents
        if isinstance(c, dict) and c.get("sector_id") == sector_id
    )
    sector_family_total_pct = direct_weight_pct + lookthrough
    return {
        "sector_id": sector_id,
        "direct_weight_pct": direct_weight_pct,
        "lookthrough_etf_weight_pct": lookthrough,
        "sector_family_total_pct": sector_family_total_pct,
        "gate": "PASS",
    }


def residualize_factor_beta(factor_beta_raw: float, peer_sector_betas: list | None) -> tuple[float | None, str]:
    if peer_sector_betas is None:
        return factor_beta_raw, "PARTIAL_raw_beta_peer_data_missing"
    shared_variance = sum(float(p.get("shared_variance", 0.0)) for p in peer_sector_betas if isinstance(p, dict))
    return factor_beta_raw - shared_variance, "PASS"


def evaluate_leader_role(position: dict) -> dict:
    required_fields = [
        "relative_strength_leads_sector",
        "volume_quality_confirmed",
        "above_ma60_or_reclaim_confirmed",
        "earnings_revision_status",
        "institutional_flow_status",
    ]
    current_role = position.get("current_role") or "LAGGARD"
    if any(position.get(f) is None for f in required_fields):
        return {"leader_role": current_role, "role_transition_reason": "DATA_MISSING", "role_changed": False}

    demotion = (
        (position["above_ma60_or_reclaim_confirmed"] is False and position["institutional_flow_status"] == "distribution")
        or position["earnings_revision_status"] == "negative"
        or (
            position["institutional_flow_status"] == "distribution"
            and current_role in ("CAPTAIN", "CORE_LEADER")
        )
    )
    if demotion:
        return {
            "leader_role": "DISTRIBUTION_RISK",
            "role_transition_reason": "demotion_trigger",
            "role_changed": current_role != "DISTRIBUTION_RISK",
        }

    promotion_ok = (
        position["relative_strength_leads_sector"] is True
        and position["volume_quality_confirmed"] is True
        and position["above_ma60_or_reclaim_confirmed"] is True
        and position["earnings_revision_status"] != "negative"
        and position["institutional_flow_status"] != "distribution"
    )
    if promotion_ok and current_role in PROMOTION_PATH:
        idx = PROMOTION_PATH.index(current_role)
        next_role = PROMOTION_PATH[min(idx + 1, len(PROMOTION_PATH) - 1)]
        return {
            "leader_role": next_role,
            "role_transition_reason": "promotion_requires_all_satisfied",
            "role_changed": next_role != current_role,
        }

    return {"leader_role": current_role, "role_transition_reason": "no_change", "role_changed": False}


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

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

    rows = []
    for position in positions:
        if not isinstance(position, dict):
            continue
        exposure = sector_exposure(position)
        leader = evaluate_leader_role(position)
        beta_residualized, beta_status = residualize_factor_beta(
            float(position.get("factor_beta_raw") or 0.0), position.get("peer_sector_betas")
        )
        rows.append({**exposure, **leader, "factor_beta_residualized": beta_residualized, "beta_status": beta_status})

    result = {
        "formula_id": "SECTOR_EXPOSURE_GRAPH_V1",
        "gate": "PASS" if rows else "DATA_MISSING",
        "rows": rows,
        "source_paths": [str(Path(args.positions))],
    }
    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())