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 okx_codex_trader.okx_client import OkxClient from scripts import build_eth_focused_portfolio_signal_intent as intent from scripts import explore_ultrashort as explore ROOT = Path(__file__).resolve().parents[1] STATE_DIR = ROOT / "var" / "eth-focused-portfolio" SYMBOLS = ("BTC-USDT-SWAP", "ETH-USDT-SWAP") LATEST_REFRESH_LIMIT = 100 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 merge_latest_candles(client: OkxClient, symbol: str, bar: str) -> list[object]: cached, history_exhausted = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, bar) merged = {candle.ts: candle for candle in cached} for candle in client.get_candles(symbol, bar, LATEST_REFRESH_LIMIT): merged[candle.ts] = candle candles = sorted(merged.values(), key=lambda candle: candle.ts) explore.save_cached_candles(explore.CANDLE_CACHE_DIR, symbol, bar, candles, history_exhausted) return candles def cached_candles(symbol: str, bar: str) -> list[object]: candles, _history_exhausted = explore.load_cached_candles(explore.CANDLE_CACHE_DIR, symbol, bar) return candles def candle_row(candles: list[object]) -> dict[str, object]: return { "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"), } def refresh_candles(client: OkxClient) -> dict[str, object]: rows = {} for symbol in SYMBOLS: rows[symbol] = { "15m": candle_row(cached_candles(symbol, "15m")), "5m": candle_row(merge_latest_candles(client, symbol, "5m")), } return rows def run_once(state_dir: Path) -> dict[str, object]: state_dir.mkdir(parents=True, exist_ok=True) candles = refresh_candles(OkxClient()) 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.JSON_REPORT.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8") intent.MARKDOWN_REPORT.write_text(intent.markdown_report(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 main() -> int: parser = argparse.ArgumentParser(description="Run ETH-focused portfolio 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": "eth_focused_portfolio_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())