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())