okx-codex-trader/scripts/build_eth_conservative_portfolio_report.py

#!/usr/bin/env python3
from __future__ import annotations

import csv
import json
from pathlib import Path
from typing import Any


ROOT = Path(__file__).resolve().parents[1]
REPORT_DIR = ROOT / "reports" / "eth-exploration"
MD_OUT = REPORT_DIR / "eth-conservative-portfolio-final.md"
JSON_OUT = REPORT_DIR / "eth-conservative-portfolio-final.json"


def read_csv(name: str) -> list[dict[str, str]]:
    with (REPORT_DIR / name).open(newline="", encoding="utf-8") as handle:
        return list(csv.DictReader(handle))


def f(row: dict[str, str], key: str) -> float:
    value = row.get(key, "")
    return float(value) if value not in ("", None) else 0.0


def pct(value: float) -> str:
    return f"{value * 100:.2f}%"


def num(value: float) -> str:
    return f"{value:.3f}"


def pick(rows: list[dict[str, str]], **where: str) -> list[dict[str, str]]:
    return [row for row in rows if all(row.get(key) == value for key, value in where.items())]


def public_metrics(row: dict[str, str]) -> dict[str, Any]:
    keys = [
        "cost_model",
        "scope",
        "trades",
        "net_total_return",
        "net_annualized_return",
        "net_max_drawdown",
        "net_calmar",
        "worst_month",
        "worst_month_return",
        "min_horizon_total_return",
        "max_horizon_drawdown",
    ]
    out: dict[str, Any] = {}
    for key in keys:
        if key not in row:
            continue
        value = row[key]
        if key in {
            "trades",
            "net_total_return",
            "net_annualized_return",
            "net_max_drawdown",
            "net_calmar",
            "worst_month_return",
            "min_horizon_total_return",
            "max_horizon_drawdown",
        }:
            out[key] = f(row, key)
        else:
            out[key] = value
    return out


def candidate_from_portfolio(priority: int, title: str, row: dict[str, str], decision: str, next_step: str, real_live_now: bool) -> dict[str, Any]:
    return {
        "priority": priority,
        "title": title,
        "name": row["portfolio"],
        "status": "candidate",
        "real_live_now": real_live_now,
        "needs_forward_or_demo_live": True,
        "minimum_next_step": next_step,
        "decision": decision,
        "metrics": public_metrics(row),
        "legs": row["legs"].split(";"),
        "weights": row["weights"].split(";"),
    }


def candidate_from_bb(priority: int, row: dict[str, str]) -> dict[str, Any]:
    return {
        "priority": priority,
        "title": "BB squeeze risk candidate",
        "name": row["name"],
        "status": "watchlist",
        "real_live_now": False,
        "needs_forward_or_demo_live": True,
        "minimum_next_step": "Run shadow/demo signal intent logging only; require fresh forward trades before any capital allocation because the acceptable-risk versions have few trades.",
        "decision": "Keep as secondary watchlist, not primary allocation.",
        "metrics": public_metrics(row),
    }


def build_payload() -> dict[str, Any]:
    portfolios = read_csv("eth-focused-portfolio-conservative-qualified.csv")
    bb = read_csv("eth-bb-squeeze-risk-10y-summary.csv")
    twap_cons = read_csv("eth-twap-conservative-ranked.csv")
    taker_top = read_csv("eth-twap-taker-entry-top15.csv")

    maker_qualified = pick(portfolios, cost_model="maker_taker", scope="all_legs")
    no_maker = pick(portfolios, cost_model="maker_taker", scope="no_maker_dependent")
    all_legs_low_dd = maker_qualified[0]
    all_legs_high_return = sorted(maker_qualified, key=lambda row: f(row, "net_annualized_return"), reverse=True)[0]
    no_maker_low_dd = no_maker[0]

    bb_primary = [
        row
        for row in bb
        if row.get("cost") == "maker_taker"
        and f(row, "net_max_drawdown") <= 0.45
        and f(row, "worst_month_return") >= -0.25
        and f(row, "net_calmar") > 1.0
    ]
    bb_primary.sort(key=lambda row: (f(row, "net_calmar"), f(row, "net_annualized_return")), reverse=True)

    twap_nearest = twap_cons[0]
    candidates = [
        candidate_from_portfolio(
            1,
            "Lowest-drawdown ETH-focused conservative portfolio",
            all_legs_low_dd,
            "Primary next item to watch in paper/demo. It is the cleanest qualified portfolio by conservative sort, but it contains a maker-dependent TWAP leg, so real funds should wait for live fill evidence.",
            "Run quasi-live read-only/order-intent tracking for all legs and record per-leg signal, fill/miss, slippage, and portfolio equity for at least the next signal cycle set.",
            False,
        ),
        candidate_from_portfolio(
            2,
            "Simpler no-maker-dependent conservative portfolio",
            no_maker_low_dd,
            "Best fallback if maker-fill uncertainty is treated as disqualifying. It keeps qualified portfolio behavior without the robust TWAP maker-dependent leg.",
            "Shadow the two legs side by side with the primary portfolio and compare realized signal overlap and drawdown path.",
            False,
        ),
        candidate_from_portfolio(
            3,
            "Highest-return qualified conservative portfolio",
            all_legs_high_return,
            "Return-oriented qualified variant. It is less conservative than priority 1 because drawdown is materially higher.",
            "Keep as a benchmark portfolio in the same quasi-live tracker; do not allocate before it beats priority 1 on realized drawdown-adjusted behavior.",
            False,
        ),
    ]
    if bb_primary:
        candidates.append(candidate_from_bb(4, bb_primary[0]))

    rejected = [
        {
            "name": "ETH robust TWAP standalone under conservative maker assumptions",
            "status": "do_not_trade_standalone",
            "reason": "Independent validation matched the closed-trade report, but conservative maker fill/slippage assumptions had no qualified candidate with all 3y/1y/6m/3m horizons positive.",
            "nearest_miss": {
                "name": twap_nearest["name"],
                "trades": f(twap_nearest, "trades"),
                "net_total_return": f(twap_nearest, "net_total_return"),
                "net_annualized_return": f(twap_nearest, "net_annualized_return"),
                "net_max_drawdown": f(twap_nearest, "net_max_drawdown"),
                "min_horizon_total_return": f(twap_nearest, "min_horizon_total_return"),
                "worst_365_total_return": f(twap_nearest, "worst_365_total_return"),
            },
            "minimum_next_step": "Use only as a portfolio leg in shadow/demo tracking until actual maker fill and slippage data contradicts the conservative stress result.",
        },
        {
            "name": "ETH taker-entry TWAP",
            "status": "rejected",
            "reason": "The taker-entry search produced no eligible taker_taker candidate with positive Calmar across 3y/1y/6m/3m.",
            "eligible_rows": len(taker_top),
            "minimum_next_step": "No live work. Drop from the next real/paper-live shortlist.",
        },
    ]

    return {
        "report": "eth-conservative-portfolio-final",
        "generated_from_existing_outputs_only": True,
        "source_files": [
            "eth-focused-portfolio-conservative-qualified.csv",
            "eth-focused-portfolio-conservative-report.md",
            "eth-twap-conservative-summary.md",
            "eth-robust-twap-validation-summary.md",
            "eth-robust-twap-fill-slippage-summary.md",
            "eth-twap-taker-entry-summary.md",
            "eth-bb-squeeze-risk-10y-report.md",
            "eth-signal-intent-readonly.md",
            "eth-signal-intent-readonly.json",
        ],
        "topline_decision": "Watch the conservative portfolio layer next, not the standalone ETH TWAP. Use quasi-live/demo read-only intent first; real funds are not the minimum next step.",
        "candidates": candidates,
        "rejected_or_deprioritized": rejected,
        "current_signal_intent": {
            "completed": True,
            "latest_confirmed_candle_utc": "2026-04-29 17:00:00",
            "signal": False,
            "orders_produced": 0,
        },
    }


def md_table(rows: list[list[str]]) -> str:
    header = rows[0]
    body = rows[1:]
    lines = [
        "| " + " | ".join(header) + " |",
        "| " + " | ".join(["---"] * len(header)) + " |",
    ]
    lines.extend("| " + " | ".join(row) + " |" for row in body)
    return "\n".join(lines)


def render_markdown(payload: dict[str, Any]) -> str:
    rows = [["Priority", "Candidate", "Return", "Ann.", "MDD", "Worst month", "Risk", "Real live now", "Minimum next step"]]
    for item in payload["candidates"]:
        m = item["metrics"]
        rows.append(
            [
                str(item["priority"]),
                item["title"],
                pct(float(m.get("net_total_return", 0.0))),
                pct(float(m.get("net_annualized_return", 0.0))),
                pct(float(m.get("net_max_drawdown", 0.0))),
                pct(float(m.get("worst_month_return", 0.0))) if "worst_month_return" in m else "",
                num(float(m.get("net_calmar", 0.0))),
                "No" if not item["real_live_now"] else "Yes",
                item["minimum_next_step"],
            ]
        )

    rejected_rows = [["Item", "Status", "Key reason", "Minimum next step"]]
    for item in payload["rejected_or_deprioritized"]:
        rejected_rows.append([item["name"], item["status"], item["reason"], item["minimum_next_step"]])

    lines = [
        "# ETH conservative portfolio final decision report",
        "",
        "This report only consolidates existing ETH exploration outputs. It does not run a new search.",
        "",
        "## Topline",
        "",
        payload["topline_decision"],
        "",
        "The next thing to watch is the portfolio layer: qualified ETH-focused conservative portfolios exist, while standalone ETH TWAP is not stable enough under conservative maker-fill assumptions and taker-entry TWAP is rejected.",
        "",
        "## Recommended priority",
        "",
        md_table(rows),
        "",
        "## Candidate notes",
        "",
    ]
    for item in payload["candidates"]:
        lines.extend(
            [
                f"### P{item['priority']} {item['title']}",
                "",
                f"- Name: `{item['name']}`",
                f"- Decision: {item['decision']}",
                f"- Needs live/quasi-live evidence: {'Yes' if item['needs_forward_or_demo_live'] else 'No'}",
                f"- Real funds now: {'Yes' if item['real_live_now'] else 'No'}",
                "",
            ]
        )
        if "legs" in item:
            lines.append("Legs and weights:")
            for leg, weight in zip(item["legs"], item["weights"]):
                lines.append(f"- `{leg}` at `{weight.split('=')[-1]}`")
            lines.append("")

    lines.extend(
        [
            "## Deprioritized or rejected",
            "",
            md_table(rejected_rows),
            "",
            "## Signal intent status",
            "",
            "Readonly signal intent was completed. Latest confirmed candle was `2026-04-29 17:00:00 UTC`; signal was false and no order intent was produced.",
            "",
            "## Source outputs",
            "",
        ]
    )
    lines.extend(f"- `{name}`" for name in payload["source_files"])
    lines.append("")
    return "\n".join(lines)


def main() -> int:
    payload = build_payload()
    JSON_OUT.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    MD_OUT.write_text(render_markdown(payload), encoding="utf-8")
    print(MD_OUT)
    print(JSON_OUT)
    return 0


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