#!/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())