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") TOTAL_PATH = REPORT_DIR / "fusion-calendar-total.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") 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(TOTAL_PATH) selected = selected[ (selected["calendar_weight"] <= 0.125) & (selected["full_improved_vs_no_calendar"]) & (selected["6m_improved_vs_no_calendar"]) & (selected["3m_improved_vs_no_calendar"]) ].sort_values(["score", "calmar", "total_return"], ascending=False).head(3) if selected.empty: raise ValueError("no calendar fusion observation candidates") rows = [] for label, (_, row) in zip(["A", "B", "C"], selected.iterrows(), strict=True): rows.append( { "label": label, "name": str(row["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/fusion-calendar-total.csv", "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())