okx-codex-trader/scripts/run_eth_focused_portfolio_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 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())