okx-codex-trader/scripts/run_calendar_fusion_observer.py

from __future__ import annotations

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

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

from scripts import build_calendar_fusion_observation_intent as intent
from scripts import explore_ultrashort as explore


ROOT = Path(__file__).resolve().parents[1]
STATE_DIR = ROOT / "var" / "calendar-fusion"
SYMBOLS = ("BTC-USDT-SWAP", "ETH-USDT-SWAP")
BAR = "15m"


def now_iso() -> str:
    return datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")


def append_jsonl(path: Path, payload: dict[str, object]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as handle:
        handle.write(json.dumps(payload, sort_keys=True, separators=(",", ":")) + "\n")


def run_once(state_dir: Path) -> dict[str, object]:
    state_dir.mkdir(parents=True, exist_ok=True)
    candles = cached_candle_status()
    payload = intent.build_payload()
    payload["created_at"] = now_iso()
    payload["orders_submitted"] = 0
    payload["candles"] = candles
    intent.REPORT_DIR.mkdir(parents=True, exist_ok=True)
    intent.OUTPUT_JSON.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    intent.OUTPUT_MD.write_text(intent.markdown(payload), encoding="utf-8")
    (state_dir / "heartbeat.json").write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    append_jsonl(state_dir / "observer-events.jsonl", payload)
    return payload


def cached_candle_status() -> dict[str, object]:
    rows = {}
    for symbol in SYMBOLS:
        candles, _history_exhausted = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, BAR)
        rows[symbol] = {
            "rows": len(candles),
            "first_ts": candles[0].ts if candles else None,
            "last_ts": candles[-1].ts if candles else None,
            "last_time": "" if not candles else datetime.fromtimestamp(candles[-1].ts / 1000, UTC).isoformat().replace("+00:00", "Z"),
        }
    return rows


def main() -> int:
    parser = argparse.ArgumentParser(description="Run calendar-fusion read-only observer.")
    parser.add_argument("--state-dir", type=Path, default=STATE_DIR)
    parser.add_argument("--interval-seconds", type=int, default=300)
    parser.add_argument("--once", action="store_true")
    args = parser.parse_args()
    while True:
        try:
            payload = run_once(args.state_dir)
            print(json.dumps(payload, indent=2, sort_keys=True))
        except Exception as exc:
            error = {"created_at": now_iso(), "mode": "calendar_fusion_readonly_observer", "orders_submitted": 0, "error": str(exc)}
            append_jsonl(args.state_dir / "observer-events.jsonl", error)
            print(json.dumps(error, indent=2, sort_keys=True), file=sys.stderr)
            if args.once:
                return 1
        if args.once:
            return 0
        time.sleep(args.interval_seconds)


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