okx-codex-trader/scripts/build_high_frequency_portfolio_observation_intent.py

from __future__ import annotations

import json
import sys
from datetime import UTC, datetime
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from scripts import explore_ultrashort as explore
from scripts import search_cross_symbol_high_frequency_portfolios as hf


REPORT_DIR = Path("reports/ultrashort")
OUTPUT_JSON = REPORT_DIR / "high-frequency-portfolio-observation-intent.json"
OUTPUT_MD = REPORT_DIR / "high-frequency-portfolio-observation-intent.md"
PORTFOLIO = "risk-3-hf00486"
QUALIFIED_PATH = REPORT_DIR / "high-frequency-portfolio-qualified.csv"
YEARS = 10.0


def iso_from_ms(ts: int) -> str:
    return datetime.fromtimestamp(ts / 1000, UTC).isoformat().replace("+00:00", "Z")


def load_portfolio() -> tuple[list[str], dict[str, float], dict[str, object]]:
    row = pd.read_csv(QUALIFIED_PATH).set_index("portfolio").loc[PORTFOLIO]
    weights = {str(key): float(value) for key, value in json.loads(str(row["weights_json"])).items()}
    return str(row["legs"]).split(";"), weights, row.to_dict()


def run_leg(leg: hf.LegSpec) -> dict[str, object]:
    data = {}
    for symbol in ("BTC-USDT-SWAP", "ETH-USDT-SWAP") if leg.pair else (leg.symbol,):
        data[(symbol, leg.bar)] = hf.load_candles(symbol, leg.bar, YEARS)
    result = leg.run(data)
    latest_candle = result.candles[-1]
    latest_entry = result.entries[-1] if result.entries else None
    latest_exit = result.exits[-1] if result.exits else None
    position = result.open_position
    return {
        "leg": leg.key,
        "symbol": leg.symbol,
        "family": leg.family,
        "bar": leg.bar,
        "latest_candle_time": iso_from_ms(latest_candle.ts),
        "trade_count": result.trade_count,
        "open_position": None
        if position is None
        else {
            "side": str(position["side"]),
            "entry_time": iso_from_ms(int(position["entry_time"])),
            "entry_price": float(position["entry_price"]),
        },
        "latest_entry": None
        if latest_entry is None
        else {
            "time": iso_from_ms(int(latest_entry["ts"])),
            "side": str(latest_entry["side"]),
            "price": float(latest_entry["price"]),
        },
        "latest_exit": None
        if latest_exit is None
        else {
            "time": iso_from_ms(int(latest_exit["ts"])),
            "side": str(latest_exit["side"]),
            "price": float(latest_exit["price"]),
        },
    }


def build_payload() -> dict[str, object]:
    leg_names, weights, metrics = load_portfolio()
    specs = {leg.key: leg for leg in hf.build_legs()}
    legs = [run_leg(specs[name]) for name in leg_names]
    active = [leg for leg in legs if leg["open_position"] is not None]
    long_unit = sum(weights[str(leg["leg"])] for leg in active if leg["open_position"]["side"] == "long")
    short_unit = sum(weights[str(leg["leg"])] for leg in active if leg["open_position"]["side"] == "short")
    return {
        "mode": "readonly_observation_intent",
        "created_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
        "submitted_orders": 0,
        "private_key_required": False,
        "strategy_family": "cross_symbol_high_frequency_portfolio",
        "portfolio": PORTFOLIO,
        "source_report": "reports/ultrashort/high-frequency-portfolio-report.md",
        "risk_limits": {
            "no_order_submission": True,
            "no_cancel_submission": True,
            "blocked_for_live_trading": True,
            "reason": "high-frequency portfolio requires read-only observation before any live promotion",
        },
        "portfolio_metrics": {
            "total_return": float(metrics["total_return"]),
            "annualized_return": float(metrics["annualized_return"]),
            "max_drawdown": float(metrics["max_drawdown"]),
            "calmar": float(metrics["calmar"]),
            "trades_per_month": float(metrics["trades_per_month"]),
            "ret_1y": float(metrics["ret_1y"]),
            "ret_6m": float(metrics["ret_6m"]),
            "ret_3m": float(metrics["ret_3m"]),
        },
        "weights": weights,
        "legs": legs,
        "target": {
            "long_unit": long_unit,
            "short_unit": short_unit,
            "net_unit": long_unit - short_unit,
            "active_legs": [str(leg["leg"]) for leg in active],
        },
    }


def markdown(payload: dict[str, object]) -> str:
    lines = [
        "# High-frequency Portfolio Observation Intent",
        "",
        "Read-only observation intent. No order or cancel request was submitted.",
        "",
        f"- Portfolio: `{payload['portfolio']}`",
        f"- Net unit: `{payload['target']['net_unit']:.6f}`",
        f"- Long unit: `{payload['target']['long_unit']:.6f}`",
        f"- Short unit: `{payload['target']['short_unit']:.6f}`",
        "",
        "| leg | weight | open_side | latest_candle | latest_entry | latest_exit |",
        "| --- | ---: | --- | --- | --- | --- |",
    ]
    weights = payload["weights"]
    for leg in payload["legs"]:
        position = leg["open_position"]
        lines.append(
            f"| `{leg['leg']}` | {weights[leg['leg']]:.6f} | {position['side'] if position else 'flat'} | "
            f"{leg['latest_candle_time']} | {leg['latest_entry']['time'] if leg['latest_entry'] else ''} | "
            f"{leg['latest_exit']['time'] if leg['latest_exit'] else ''} |"
        )
    lines.extend(["", "## Intent JSON", "", "```json", json.dumps(payload, indent=2, sort_keys=True), "```", ""])
    return "\n".join(lines)


def main() -> int:
    payload = build_payload()
    REPORT_DIR.mkdir(parents=True, exist_ok=True)
    OUTPUT_JSON.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    OUTPUT_MD.write_text(markdown(payload), encoding="utf-8")
    print(json.dumps(payload, indent=2, sort_keys=True))
    return 0


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