okx-codex-trader/scripts/build_eth_focused_portfolio_signal_intent.py

from __future__ import annotations

import argparse
import csv
import json
import math
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path


ROOT = Path(__file__).resolve().parents[1]
CANDLE_DIR = ROOT / "data" / "okx-candles"
REPORT_DIR = ROOT / "reports" / "eth-exploration"
JSON_REPORT = REPORT_DIR / "eth-focused-portfolio-signal-intent.json"
MARKDOWN_REPORT = REPORT_DIR / "eth-focused-portfolio-signal-intent.md"
ETH = "ETH-USDT-SWAP"
BTC = "BTC-USDT-SWAP"
LEVERAGE = 3


@dataclass(frozen=True)
class Candle:
    ts: int
    open: float
    high: float
    low: float
    close: float
    volume: float


@dataclass(frozen=True)
class LegSpec:
    leg_id: str
    family: str
    bar: str
    weight: float
    params: dict[str, float | int]


LEGS = (
    LegSpec(
        leg_id="eth_btc_rsi_filter_15m",
        family="eth_btc_rsi_filter",
        bar="15m",
        weight=0.80314757,
        params={
            "eth_trend_sma": 50,
            "eth_rsi_threshold": 3.0,
            "eth_exit_rsi": 55.0,
            "btc_trend_sma": 480,
            "btc_momentum_lookback": 240,
            "btc_min_momentum": 0.0,
        },
    ),
    LegSpec(
        leg_id="btc_lead_eth_lag_15m",
        family="btc_lead_eth_lag",
        bar="15m",
        weight=0.09459139,
        params={
            "lead_lookback": 8,
            "btc_return_threshold": 0.018,
            "lag_gap": 0.006,
            "max_hold_bars": 8,
            "stop_loss_pct": 0.006,
            "take_profit_pct": 0.018,
        },
    ),
    LegSpec(
        leg_id="btc_lead_eth_lag_5m",
        family="btc_lead_eth_lag",
        bar="5m",
        weight=0.10226104,
        params={
            "lead_lookback": 16,
            "btc_return_threshold": 0.012,
            "lag_gap": 0.006,
            "max_hold_bars": 8,
            "stop_loss_pct": 0.006,
            "take_profit_pct": 0.018,
        },
    ),
)


def iso_text(ts: int) -> str:
    return datetime.fromtimestamp(ts / 1000, UTC).isoformat().replace("+00:00", "Z")


def load_candles(symbol: str, bar: str) -> list[Candle]:
    path = CANDLE_DIR / symbol / f"{bar}.csv"
    candles: list[Candle] = []
    with path.open("r", encoding="utf-8", newline="") as handle:
        for row in csv.DictReader(handle):
            candles.append(
                Candle(
                    ts=int(row["ts"]),
                    open=float(row["open"]),
                    high=float(row["high"]),
                    low=float(row["low"]),
                    close=float(row["close"]),
                    volume=float(row["volume"]),
                )
            )
    return sorted(candles, key=lambda candle: candle.ts)


def align_pair(eth: list[Candle], btc: list[Candle]) -> tuple[list[Candle], list[Candle]]:
    btc_by_ts = {candle.ts: candle for candle in btc}
    eth_aligned: list[Candle] = []
    btc_aligned: list[Candle] = []
    for candle in eth:
        btc_candle = btc_by_ts.get(candle.ts)
        if btc_candle is not None:
            eth_aligned.append(candle)
            btc_aligned.append(btc_candle)
    return eth_aligned, btc_aligned


def sma(values: list[float], length: int, index: int) -> float:
    if index + 1 < length:
        return math.nan
    return sum(values[index + 1 - length : index + 1]) / length


def rsi(values: list[float], length: int) -> list[float]:
    output = [math.nan] * len(values)
    if len(values) <= length:
        return output
    gains = [0.0]
    losses = [0.0]
    for previous, current in zip(values, values[1:]):
        delta = current - previous
        gains.append(max(delta, 0.0))
        losses.append(max(-delta, 0.0))
    average_gain = sum(gains[1 : length + 1]) / length
    average_loss = sum(losses[1 : length + 1]) / length
    for index in range(length, len(values)):
        if index > length:
            average_gain = ((average_gain * (length - 1)) + gains[index]) / length
            average_loss = ((average_loss * (length - 1)) + losses[index]) / length
        if average_loss == 0.0:
            output[index] = 100.0 if average_gain > 0.0 else 50.0
        else:
            relative_strength = average_gain / average_loss
            output[index] = 100.0 - (100.0 / (1.0 + relative_strength))
    return output


def eth_btc_rsi_filter_signal(spec: LegSpec, eth: list[Candle], btc: list[Candle], index: int) -> dict[str, object]:
    eth_closes = [candle.close for candle in eth]
    btc_closes = [candle.close for candle in btc]
    eth_trend_sma = int(spec.params["eth_trend_sma"])
    btc_trend_sma = int(spec.params["btc_trend_sma"])
    btc_momentum_lookback = int(spec.params["btc_momentum_lookback"])
    eth_trend = sma(eth_closes, eth_trend_sma, index)
    btc_trend = sma(btc_closes, btc_trend_sma, index)
    eth_rsi = rsi(eth_closes[: index + 1], 2)[-1]
    btc_momentum = btc[index].close / btc[index - btc_momentum_lookback].close - 1.0
    btc_risk_on = btc[index].close > btc_trend and btc_momentum >= float(spec.params["btc_min_momentum"])
    eth_pullback = eth[index].close > eth_trend and eth_rsi <= float(spec.params["eth_rsi_threshold"])
    signal = btc_risk_on and eth_pullback
    exit_signal = eth_rsi >= float(spec.params["eth_exit_rsi"]) or btc[index].close < btc_trend
    return {
        "signal": signal,
        "entry_rule": "btc_close > btc_sma and btc_momentum >= minimum and eth_close > eth_sma and eth_rsi2 <= threshold",
        "exit_signal": exit_signal,
        "indicators": {
            "eth_close": eth[index].close,
            "eth_sma": eth_trend,
            "eth_rsi2": eth_rsi,
            "btc_close": btc[index].close,
            "btc_sma": btc_trend,
            "btc_momentum": btc_momentum,
        },
    }


def btc_lead_eth_lag_signal(spec: LegSpec, eth: list[Candle], btc: list[Candle], index: int) -> dict[str, object]:
    lead_lookback = int(spec.params["lead_lookback"])
    btc_return = btc[index].close / btc[index - lead_lookback].close - 1.0
    eth_return = eth[index].close / eth[index - lead_lookback].close - 1.0
    return_gap = btc_return - eth_return
    signal = btc_return >= float(spec.params["btc_return_threshold"]) and return_gap >= float(spec.params["lag_gap"])
    return {
        "signal": signal,
        "entry_rule": "btc_return >= threshold and btc_return - eth_return >= lag_gap",
        "exit_signal": False,
        "indicators": {
            "eth_close": eth[index].close,
            "btc_close": btc[index].close,
            "btc_return": btc_return,
            "eth_return": eth_return,
            "return_gap": return_gap,
        },
    }


def evaluate_leg(spec: LegSpec, data: dict[tuple[str, str], list[Candle]]) -> dict[str, object]:
    eth, btc = align_pair(data[(ETH, spec.bar)], data[(BTC, spec.bar)])
    index = len(eth) - 2
    if spec.family == "eth_btc_rsi_filter":
        decision = eth_btc_rsi_filter_signal(spec, eth, btc, index)
    else:
        decision = btc_lead_eth_lag_signal(spec, eth, btc, index)
    stop_loss_pct = spec.params.get("stop_loss_pct")
    take_profit_pct = spec.params.get("take_profit_pct")
    return {
        "leg_id": spec.leg_id,
        "family": spec.family,
        "symbol": ETH,
        "bar": spec.bar,
        "decision_candle_ts": eth[index].ts,
        "decision_candle_time": iso_text(eth[index].ts),
        "latest_local_candle_ts": eth[-1].ts,
        "latest_local_candle_time": iso_text(eth[-1].ts),
        "suggested_weight": spec.weight,
        "signal": decision["signal"],
        "needs_order": bool(decision["signal"]),
        "needs_cancel": False,
        "dry_run_action": "would_open_long" if decision["signal"] else "hold",
        "risk_limits": {
            "leverage": LEVERAGE,
            "max_weight": spec.weight,
            "stop_loss_pct": stop_loss_pct,
            "take_profit_pct": take_profit_pct,
            "max_hold_bars": spec.params.get("max_hold_bars"),
        },
        "params": spec.params,
        "entry_rule": decision["entry_rule"],
        "exit_signal": decision["exit_signal"],
        "indicators": decision["indicators"],
    }


def build_payload() -> dict[str, object]:
    bars = sorted({leg.bar for leg in LEGS})
    data = {(symbol, bar): load_candles(symbol, bar) for symbol in (ETH, BTC) for bar in bars}
    legs = [evaluate_leg(spec, data) for spec in LEGS]
    active_weight = sum(float(leg["suggested_weight"]) for leg in legs if leg["signal"])
    return {
        "mode": "dry_run_readonly_portfolio_signal_intent",
        "created_at": datetime.now(UTC).isoformat(timespec="seconds").replace("+00:00", "Z"),
        "submitted_orders": 0,
        "order_client": None,
        "private_key_required": False,
        "portfolio": {
            "name": "eth_focused_conservative_signal_intent",
            "symbol": ETH,
            "direction": "long_only",
            "basis": "ETH/BTC RSI filter + BTC lead ETH lag 5m/15m",
            "leverage": LEVERAGE,
            "active_signal_count": sum(1 for leg in legs if leg["signal"]),
            "active_suggested_weight": active_weight,
            "needs_order": any(bool(leg["needs_order"]) for leg in legs),
            "needs_cancel": any(bool(leg["needs_cancel"]) for leg in legs),
            "dry_run_action": "would_open_or_rebalance_long" if active_weight > 0.0 else "hold",
        },
        "risk_limits": {
            "portfolio_max_gross_weight": 1.0,
            "leg_weights_sum": sum(spec.weight for spec in LEGS),
            "no_order_submission": True,
            "no_cancel_submission": True,
            "no_position_state_assumed": True,
            "execution": "intent_only",
        },
        "legs": legs,
    }


def markdown_report(payload: dict[str, object]) -> str:
    lines = [
        "# ETH-focused portfolio signal intent",
        "",
        "Dry-run only. No order or cancel request was submitted.",
        "",
        "## Portfolio",
        "",
        f"- Created at: `{payload['created_at']}`",
        f"- Direction: `{payload['portfolio']['direction']}`",
        f"- Active signal count: `{payload['portfolio']['active_signal_count']}`",
        f"- Active suggested weight: `{payload['portfolio']['active_suggested_weight']:.8f}`",
        f"- Needs order: `{payload['portfolio']['needs_order']}`",
        f"- Needs cancel: `{payload['portfolio']['needs_cancel']}`",
        "",
        "## Legs",
        "",
        "| Leg | Bar | Signal | Weight | Action | Decision candle |",
        "| --- | --- | --- | --- | --- | --- |",
    ]
    for leg in payload["legs"]:
        lines.append(
            f"| `{leg['leg_id']}` | `{leg['bar']}` | `{leg['signal']}` | `{leg['suggested_weight']:.8f}` | `{leg['dry_run_action']}` | `{leg['decision_candle_time']}` |"
        )
    lines.extend(
        [
            "",
            "## Intent JSON",
            "",
            "```json",
            json.dumps(payload, indent=2, sort_keys=True),
            "```",
        ]
    )
    return "\n".join(lines) + "\n"


def main() -> int:
    parser = argparse.ArgumentParser(description="Build a read-only ETH-focused portfolio signal/intent payload.")
    parser.add_argument("--no-write", action="store_true")
    args = parser.parse_args()
    payload = build_payload()
    if not args.no_write:
        REPORT_DIR.mkdir(parents=True, exist_ok=True)
        JSON_REPORT.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
        MARKDOWN_REPORT.write_text(markdown_report(payload), encoding="utf-8")
    print(json.dumps(payload, indent=2, sort_keys=True))
    return 0


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