okx-codex-trader/scripts/build_calendar_fusion_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 search_eth_btc_calendar_carry as carry


REPORT_DIR = Path("reports/long-short-fusion")
SELECTED_PATH = REPORT_DIR / "fusion-calendar-selected.csv"
OUTPUT_JSON = REPORT_DIR / "calendar-fusion-observation-intent.json"
OUTPUT_MD = REPORT_DIR / "calendar-fusion-observation-intent.md"
CANDIDATE_SPEC = carry.Spec("ETH-USDT-SWAP", "1h", "long", 14, "weekend", 8, "calm")
CANDIDATES = {
    "A": "fusion-cal-lr_riskoff00-l1.20-brs0.08-eth4hs0.10-btc4hs0.06-eg0.00-bg0.00-cal0.125",
    "B": "fusion-cal-lr_riskoff00-l1.20-brs0.08-eth4hs0.00-btc4hs0.04-eg0.12-bg0.00-cal0.125",
    "C": "fusion-cal-lr_riskoff00-l1.20-brs0.08-eth4hs0.00-btc4hs0.04-eg0.12-bg0.00-cal0.075",
}
BAR_DELTAS = {"1h": pd.Timedelta(hours=1), "4h": pd.Timedelta(hours=4)}


def iso(ts: pd.Timestamp) -> str:
    return ts.isoformat().replace("+00:00", "Z")


def calendar_state() -> dict[str, object]:
    frame = carry.resample(carry.load_frame(CANDIDATE_SPEC.symbol), CANDIDATE_SPEC.bar)
    returns = frame["close"].pct_change()
    vol = returns.rolling(96).std(ddof=0)
    vol_rank = vol.rolling(720).rank(pct=True)
    allowed_weekdays = carry.WEEKDAY_SETS[CANDIDATE_SPEC.weekdays]
    signals = (
        (frame.index.hour == CANDIDATE_SPEC.hour)
        & pd.Series(frame.index.weekday, index=frame.index).isin(allowed_weekdays)
        & (vol_rank <= 0.5)
    ).fillna(False)
    decision_index = len(frame) - 2
    decision_time = frame.index[decision_index]
    latest_signal_times = list(signals[signals & (signals.index <= decision_time)].index)
    latest_signal_time = latest_signal_times[-1] if latest_signal_times else None
    active = False
    entry_time = None
    exit_time = None
    if latest_signal_time is not None:
        signal_index = frame.index.get_loc(latest_signal_time)
        entry_index = signal_index + 1
        exit_index = entry_index + CANDIDATE_SPEC.hold
        bar_delta = BAR_DELTAS[CANDIDATE_SPEC.bar]
        entry_time = frame.index[entry_index] if entry_index < len(frame) else latest_signal_time + bar_delta
        exit_time = frame.index[exit_index] if exit_index < len(frame) else entry_time + (bar_delta * CANDIDATE_SPEC.hold)
        active = bool(entry_time <= decision_time < exit_time)
    return {
        "spec": CANDIDATE_SPEC.name,
        "decision_candle_time": iso(decision_time),
        "latest_local_candle_time": iso(frame.index[-1]),
        "entry_signal_on_decision_candle": bool(signals.iloc[decision_index]),
        "vol_rank_on_decision_candle": None if pd.isna(vol_rank.iloc[decision_index]) else float(vol_rank.iloc[decision_index]),
        "latest_signal_time": iso(latest_signal_time) if latest_signal_time is not None else None,
        "theoretical_entry_time": iso(entry_time) if entry_time is not None else None,
        "theoretical_exit_time": iso(exit_time) if exit_time is not None else None,
        "theoretical_calendar_position_active": active,
        "side": CANDIDATE_SPEC.side,
    }


def selected_candidates() -> list[dict[str, object]]:
    selected = pd.read_csv(SELECTED_PATH).set_index("name")
    rows = []
    for label, name in CANDIDATES.items():
        row = selected.loc[name]
        rows.append(
            {
                "label": label,
                "name": name,
                "calendar_weight": float(row["calendar_weight"]),
                "gross_exposure": float(row["gross_exposure"]),
                "short_exposure": float(row["short_exposure"]),
                "component_weights": {
                    "long_variant": str(row["long_variant"]),
                    "long_weight": float(row["long_weight"]),
                    "btc_risk_short_weight": float(row["btc_risk_short_weight"]),
                    "eth_4h_vol_short_weight": float(row["eth_4h_vol_short_weight"]),
                    "btc_4h_vol_short_weight": float(row["btc_4h_vol_short_weight"]),
                    "eth_4h_vol_short_gated_weight": float(row["eth_4h_vol_short_gated_weight"]),
                    "btc_4h_vol_short_gated_weight": float(row["btc_4h_vol_short_gated_weight"]),
                    "calendar_weight": float(row["calendar_weight"]),
                },
                "metrics": {
                    "full_total_return": float(row["total_return"]),
                    "full_annualized_return": float(row["annualized_return"]),
                    "full_max_drawdown": float(row["max_drawdown"]),
                    "full_calmar": float(row["calmar"]),
                    "h3y_return": float(row["h3y_return"]),
                    "h1y_return": float(row["h1y_return"]),
                    "h6m_return": float(row["h6m_return"]),
                    "h3m_return": float(row["h3m_return"]),
                },
            }
        )
    return rows


def build_payload() -> dict[str, object]:
    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": "calendar_fusion_observation",
        "source_report": "reports/long-short-fusion/calendar-fusion-observation-candidates.md",
        "risk_limits": {
            "no_order_submission": True,
            "no_cancel_submission": True,
            "blocked_for_live_trading": True,
            "reason": "candidate requires read-only observation before any live promotion",
        },
        "calendar_state": calendar_state(),
        "candidates": selected_candidates(),
    }


def markdown(payload: dict[str, object]) -> str:
    state = payload["calendar_state"]
    lines = [
        "# Calendar Fusion Observation Intent",
        "",
        "Read-only observation intent. No order or cancel request was submitted.",
        "",
        "## Calendar Leg",
        "",
        f"- Spec: `{state['spec']}`",
        f"- Decision candle: `{state['decision_candle_time']}`",
        f"- Latest local candle: `{state['latest_local_candle_time']}`",
        f"- Entry signal on decision candle: `{state['entry_signal_on_decision_candle']}`",
        f"- Theoretical active position: `{state['theoretical_calendar_position_active']}`",
        f"- Latest signal time: `{state['latest_signal_time']}`",
        f"- Theoretical entry: `{state['theoretical_entry_time']}`",
        f"- Theoretical exit: `{state['theoretical_exit_time']}`",
        "",
        "## Candidates",
        "",
        "| label | calendar_weight | gross_exposure | short_exposure | full_return | max_dd | calmar |",
        "| --- | ---: | ---: | ---: | ---: | ---: | ---: |",
    ]
    for row in payload["candidates"]:
        metrics = row["metrics"]
        lines.append(
            f"| {row['label']} | {row['calendar_weight']:.3f} | {row['gross_exposure']:.3f} | {row['short_exposure']:.3f} | "
            f"{metrics['full_total_return']:.2%} | {metrics['full_max_drawdown']:.2%} | {metrics['full_calmar']:.2f} |"
        )
    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())