#!/usr/bin/env python3
"""
Shared signal + backtest engine for the Coinbase stat-arb v3 forward test
and Pareto hill-climbing optimizer.

Strategy (arbitrated v3, 2026-10-05):
  Short front-month dated future / long perp-style future 1:1 per pair
  (BTC: BIT/BIP, XRP: XRP/XPP). Slow carry book only; overlay suspended.

Conventions:
  - All P&L in dollars on the SPREAD, never in % of a leg.
  - Basis measured executable-first: (dated_bid - perp_ask); candle mids get a
    spread haircut instead (stated assumption, sensitivity-tested).
  - Fees: dated 5 bps/side, perp 2 bps/side at institutional clips
    (the $0.20/$0.15 minimums only bind on tiny clips; noted where relevant).
  - Funding: hourly rate from the perp contract; long perp pays when positive.

Paper/research only. No order placement anywhere in this file.
"""

from dataclasses import dataclass, field
from datetime import datetime, timezone
import math

# ---------------------------------------------------------------- constants

V3_DEFAULTS = {
    "entry_C": 3.0,        # % ann. net carry to add
    "standdown_C": 1.5,    # % ann. net carry -> stand down
    "persist_bars": 1,     # consecutive bars the gate must hold (hourly data)
    "clip_usd": {"BTC": 35000.0, "XRP": 10000.0},
    "use_sessions": True,  # respect session/blackout gates
    "spread_haircut_bps": 8.0,  # deducted from mid basis (executable proxy)
}

FEE_DATED_BPS = 5.0   # per side
FEE_PERP_BPS = 2.0    # per side
ROUNDTRIP_BPS = FEE_DATED_BPS + 2 * FEE_PERP_BPS  # enter both legs + flatten perp

_COINBASE_REST = "https://api.coinbase.com/api/v3/brokerage"
_expiry_cache = {}


def contract_expiry_ts(product_id):
    """True expiry unix ts from contract metadata (never hardcode)."""
    if product_id not in _expiry_cache:
        import requests
        from datetime import datetime
        r = requests.get(f"{_COINBASE_REST}/market/products/{product_id}",
                         timeout=20)
        r.raise_for_status()
        fpd = (r.json().get("product", r.json())
               .get("future_product_details", {}) or {})
        iso = fpd.get("contract_expiry")
        if not iso:
            raise ValueError(f"no contract_expiry for {product_id}")
        dt = datetime.fromisoformat(iso.replace("Z", "+00:00"))
        _expiry_cache[product_id] = dt.timestamp()
    return _expiry_cache[product_id]

# (start_hour_et, end_hour_et, size_mult) — blackout 17-19 ET handled separately
SESSIONS = [(8, 17, 1.0), (4, 8, 1.0), (19, 24, 0.5), (0, 3, 0.5)]
BLACKOUT = [(17, 19)]  # daily; Friday 17-18 halt folded in conservatively


def et_hour(ts):
    """Hour of day in America/New_York for a unix timestamp (EDT in Oct)."""
    # Oct 2026: EDT = UTC-4. (No DST math needed inside the window.)
    return (datetime.fromtimestamp(ts, timezone.utc).hour - 4) % 24


def session_mult(ts):
    h = et_hour(ts)
    for a, b in BLACKOUT:
        if a <= h < b:
            return 0.0
    for a, b, m in SESSIONS:
        if a <= h < b:
            return m
    return 0.0


# ---------------------------------------------------------------- data

@dataclass
class Bar:
    ts: float
    dated: float      # dated future mid (or executable bid when available)
    perp: float       # perp mid (or executable ask when available)
    funding_hr: float  # perp hourly funding rate, signed (long pays if > 0)
    expiry_ts: float
    executable: bool = False  # True if dated/perp are bid/ask, not mids


@dataclass
class Position:
    pair: str
    qty_dated: float   # contracts, short => negative
    qty_perp: float    # contracts, long => positive
    notional: float    # USD notional per leg at entry
    entry_spread: float  # (dated - perp) per unit of dated at entry, in USD of dated px
    entry_ts: float
    entry_basis_bps: float


@dataclass
class Ledger:
    equity: float
    cash_pnl: float = 0.0
    funding_pnl: float = 0.0
    fees_paid: float = 0.0
    positions: dict = field(default_factory=dict)  # pair -> Position
    curve: list = field(default_factory=list)      # (ts, equity_mtm) for Sharpe/DD
    trades: list = field(default_factory=list)


# ---------------------------------------------------------------- signals

def net_carry_pct(bar: Bar, haircut_bps: float) -> float:
    """7-day-style net carry C = annualized executable basis - annualized funding."""
    hours_left = max((bar.expiry_ts - bar.ts) / 3600.0, 1.0)
    basis_bps = (bar.dated - bar.perp) / bar.perp * 1e4
    if not bar.executable:
        basis_bps -= haircut_bps
    ann_basis = basis_bps / 1e4 * (365 * 24 / hours_left) * 100.0
    ann_funding = bar.funding_hr * 24 * 365 * 100.0
    return ann_basis - ann_funding


def decide(bars_by_pair, params, ledger):
    """
    One evaluation pass over the latest bar of each pair.
    Returns list of action dicts; mutates ledger (paper fills).
    Caller is responsible for the persistence gate (consecutive bars).
    """
    actions = []
    for pair, bars in bars_by_pair.items():
        bar = bars[-1]
        C = net_carry_pct(bar, params["spread_haircut_bps"])
        in_pos = pair in ledger.positions
        sess = session_mult(bar.ts) if params["use_sessions"] else 1.0

        if not in_pos:
            if C >= params["entry_C"] and sess > 0:
                clip = params["clip_usd"][pair] * sess
                # entry friction: short dated (5 bps) + long perp (2 bps), one side each
                # contracts: dated short, perp long (contract sizes differ per root)
                ledger.fees_paid += clip * (FEE_DATED_BPS + FEE_PERP_BPS) / 1e4
                ledger.positions[pair] = Position(
                    pair=pair, qty_dated=-1.0, qty_perp=1.0,
                    notional=clip, entry_spread=bar.dated - bar.perp,
                    entry_ts=bar.ts,
                    entry_basis_bps=(bar.dated - bar.perp) / bar.perp * 1e4)
                actions.append({"ts": bar.ts, "pair": pair, "action": "ENTER",
                                "C": round(C, 2), "clip": clip,
                                "basis_bps": round((bar.dated - bar.perp) / bar.perp * 1e4, 1)})
        else:
            pos = ledger.positions[pair]
            # MTM on the spread: short spread profits when spread narrows
            spread_now = bar.dated - bar.perp
            mtm = pos.notional / bar.perp * (pos.entry_spread - spread_now)
            # funding accrual for this bar (hourly): long perp pays funding>0
            fund = -bar.funding_hr * pos.notional
            ledger.funding_pnl += fund
            exit_now = False
            reason = ""
            if C < params["standdown_C"]:
                exit_now, reason = True, f"net carry {C:.2f}% < standdown"
            elif bar.ts >= bar.expiry_ts - 3600:  # flatten 1h before expiry
                exit_now, reason = True, "expiry flatten"
            if exit_now:
                ledger.cash_pnl += mtm
                ledger.fees_paid += pos.notional * FEE_PERP_BPS / 1e4  # flatten perp
                actions.append({"ts": bar.ts, "pair": pair, "action": "EXIT",
                                "reason": reason, "mtm": round(mtm, 2)})
                del ledger.positions[pair]
    return actions


def mtm_equity(bars_by_pair, ledger):
    eq = ledger.equity + ledger.cash_pnl + ledger.funding_pnl - ledger.fees_paid
    for pair, pos in ledger.positions.items():
        bar = bars_by_pair[pair][-1]
        spread_now = bar.dated - bar.perp
        eq += pos.notional / bar.perp * (pos.entry_spread - spread_now)
    return eq


# ---------------------------------------------------------------- backtest

def backtest(bars_by_pair, funding_const, params, equity=50000.0, expiry_ts=None):
    """
    Walk hourly bars through the v3 rules. Returns (metrics, ledger).
    funding_const: {pair: hourly_rate} used when bars carry no funding series.
    """
    pairs = list(bars_by_pair.keys())
    n = min(len(b) for b in bars_by_pair.values())
    led = Ledger(equity=equity)
    persist = {p: 0 for p in pairs}

    for i in range(n):
        window = {}
        for p in pairs:
            b = bars_by_pair[p][i]
            if b.funding_hr is None:
                b = Bar(b.ts, b.dated, b.perp, funding_const[p], b.expiry_ts, b.executable)
                bars_by_pair[p][i] = b
            window[p] = bars_by_pair[p][:i + 1]

        # persistence gate: count consecutive bars with C >= entry_C per pair
        gated_window = {}
        for p in pairs:
            C = net_carry_pct(window[p][-1], params["spread_haircut_bps"])
            in_pos = p in led.positions
            if not in_pos and C >= params["entry_C"]:
                persist[p] += 1
            else:
                persist[p] = 0
            # only hand the bar to decide() when the gate has held long enough
            if in_pos or persist[p] >= params["persist_bars"]:
                gated_window[p] = window[p]
            else:
                # still need MTM/funding accrual on existing positions: none here
                pass
        # positions already open still need their per-bar accrual: run decide on
        # open positions every bar regardless of the gate
        for p in list(led.positions):
            if p not in gated_window:
                gated_window[p] = window[p]
        if gated_window:
            acts = decide(gated_window, params, led)
            led.trades.extend(acts)
        led.curve.append((window[pairs[0]][-1].ts, mtm_equity(window, led)))

    # final MTM close for metrics (positions left open are marked, not liquidated)
    final_eq = led.curve[-1][1] if led.curve else equity
    metrics = summarize(led, final_eq, equity)
    return metrics, led


def summarize(ledger, final_eq, equity):
    import statistics
    pnl = final_eq - equity
    n_days = 1.0
    if len(ledger.curve) >= 2:
        n_days = max((ledger.curve[-1][0] - ledger.curve[0][0]) / 86400.0, 1 / 24)
    apy = pnl / equity * (365.0 / n_days) * 100.0
    eqs = [e for _, e in ledger.curve]
    # daily-ish Sharpe from curve
    rets = []
    step = max(len(eqs) // max(int(n_days), 1), 1)
    for i in range(step, len(eqs), step):
        if eqs[i - step] > 0:
            rets.append((eqs[i] - eqs[i - step]) / eqs[i - step])
    sharpe = (statistics.mean(rets) / statistics.pstdev(rets) * math.sqrt(365)
              if len(rets) > 2 and statistics.pstdev(rets) > 0 else 0.0)
    peak, maxdd = eqs[0], 0.0
    for e in eqs:
        peak = max(peak, e)
        maxdd = max(maxdd, (peak - e) / peak * 100.0)
    n_trades = sum(1 for a in ledger.trades if a.get("action") == "ENTER")
    return {
        "pnl": round(pnl, 2),
        "apy_pct": round(apy, 2),
        "sharpe": round(sharpe, 2),
        "maxdd_pct": round(maxdd, 2),
        "fees": round(ledger.fees_paid, 2),
        "funding_pnl": round(ledger.funding_pnl, 2),
        "n_trades": n_trades,
        "n_days": round(n_days, 1),
    }

#!/usr/bin/env python3
"""
Forward data recorder for the Coinbase stat-arb validation program.

Records, into a local SQLite database:
  - Executable top-of-book + 3-level depth for the dated/perp pairs,
    polled every 10s via the public product_book endpoint
    (the WS l2_data stream does not complete its handshake from this network)
  - Hourly funding rates, index prices, margin rates, settlement prices (polled 60s)

This is the FIRST step of the arbitrated v3 design: record 2-3 full expiry cycles
before any sizing decision. Paper/research only — no orders are placed.

Requires: requests
    pip install requests

Tables:
  books(ts, product_id, bid_px, bid_sz, ask_px, ask_sz, bid_depth3, ask_depth3,
        spread_bps)
  funding(ts, product_id, rate)
  meta(ts, product_id, index_price, margin_intraday, margin_overnight, settlement_price)
"""

import json
import sqlite3
import threading
import time
from datetime import datetime, timezone

import requests

REST = "https://api.coinbase.com/api/v3/brokerage"
WS_URL = "wss://advanced-trade-ws.coinbase.com"  # unavailable from this network

# Roots we care about: (dated_root, perp_root). Dated leg = nearest monthly expiry.
PAIRS = [("BIT", "BIP"), ("XRP", "XPP"), ("ET", "ETP")]

DB_PATH = "stat_arb_data.db"
BOOK_POLL_S = 10  # product_book cadence (6 products -> ~36 req/min, well under limits)


# ---------------------------------------------------------------- discovery

def _detail(pid):
    r = requests.get(f"{REST}/market/products/{pid}", timeout=20)
    r.raise_for_status()
    return r.json().get("product", r.json())


def _expiry_of(detail):
    iso = (detail.get("future_product_details", {}) or {}).get("contract_expiry")
    if not iso:
        return None
    from datetime import datetime
    return datetime.fromisoformat(iso.replace("Z", "+00:00")).timestamp()


def discover_products():
    """Return {dated_root: dated_product_id, perp_root: perp_product_id}.

    Dated leg = nearest-expiry monthly contract (by true contract_expiry,
    not product-id string sort). Perp leg = the 2089-dated perp-style listing.
    """
    r = requests.get(f"{REST}/market/products",
                     params={"product_type": "FUTURE", "limit": 250}, timeout=30)
    r.raise_for_status()
    prods = [p for p in r.json().get("products", []) if not p.get("is_disabled")]
    now = time.time()
    out = {}
    for dated_root, perp_root in PAIRS:
        dated_cands = [p["product_id"] for p in prods
                       if p["product_id"].startswith(dated_root + "-")
                       and "20DEC30" not in p["product_id"]]
        best, best_exp = None, None
        for pid in dated_cands:
            try:
                exp = _expiry_of(_detail(pid))
            except Exception:
                continue
            if exp and exp > now and (best_exp is None or exp < best_exp):
                best, best_exp = pid, exp
        if best:
            out[dated_root] = best
        perp_cands = [p["product_id"] for p in prods
                      if "20DEC30" in p["product_id"]
                      and p["product_id"].startswith(perp_root)]
        if perp_cands:
            out[perp_root] = sorted(perp_cands)[0]
    return out


# ---------------------------------------------------------------- storage

SCHEMA = """
DROP TABLE IF EXISTS books;
CREATE TABLE books(
  ts REAL, product_id TEXT,
  bid_px REAL, bid_sz REAL, ask_px REAL, ask_sz REAL,
  bid_depth3 REAL, ask_depth3 REAL, spread_bps REAL);
CREATE TABLE IF NOT EXISTS funding(
  ts REAL, product_id TEXT, rate REAL);
CREATE TABLE IF NOT EXISTS meta(
  ts REAL, product_id TEXT, index_price REAL,
  margin_intraday REAL, margin_overnight REAL, settlement_price REAL);
CREATE INDEX IF NOT EXISTS idx_books ON books(product_id, ts);
CREATE INDEX IF NOT EXISTS idx_funding ON funding(product_id, ts);
"""


def open_db(path=DB_PATH):
    db = sqlite3.connect(path, check_same_thread=False)
    db.executescript(SCHEMA)
    return db


# ---------------------------------------------------------------- book polling (REST)

def poll_books(db, product_ids):
    """Every BOOK_POLL_S: executable top-of-book + 3-level depth per product."""
    while True:
        now = time.time()
        for pid in product_ids:
            try:
                r = requests.get(f"{REST}/market/product_book",
                                 params={"product_id": pid, "limit": 3},
                                 timeout=15)
                if r.status_code != 200:
                    continue
                pb = r.json().get("pricebook", {})
                bids = pb.get("bids", []) or []
                asks = pb.get("asks", []) or []
                if not bids or not asks:
                    continue
                db.execute(
                    "INSERT INTO books VALUES (?,?,?,?,?,?,?,?,?)",
                    (now, pid,
                     float(bids[0]["price"]), float(bids[0]["size"]),
                     float(asks[0]["price"]), float(asks[0]["size"]),
                     sum(float(b["size"]) for b in bids[:3]),
                     sum(float(a["size"]) for a in asks[:3]),
                     float(r.json().get("spread_bps") or 0)))
            except Exception as e:
                print("book poll error:", pid, e, flush=True)
        db.commit()
        time.sleep(BOOK_POLL_S)


# ---------------------------------------------------------------- rest poll

def poll_meta(db, product_ids):
    """Every 60s: funding rate, index price, margin rates, settlement price."""
    while True:
        now = time.time()
        for pid in product_ids:
            try:
                r = requests.get(f"{REST}/market/products/{pid}", timeout=20)
                if r.status_code != 200:
                    continue
                p = r.json().get("product", r.json())
                f = p.get("funding_rate")
                if f is not None:
                    db.execute("INSERT INTO funding VALUES (?,?,?)",
                               (now, pid, float(f)))
                db.execute("INSERT INTO meta VALUES (?,?,?,?,?,?)", (
                    now, pid,
                    _f(p.get("index_price")),
                    _f(p.get("intraday_margin_rate")),
                    _f(p.get("overnight_margin_rate")),
                    _f(p.get("settlement_price") or p.get("price")),
                ))
            except Exception as e:
                print("meta poll error:", pid, e, flush=True)
        db.commit()
        time.sleep(60)


def _f(v):
    try:
        return float(v)
    except (TypeError, ValueError):
        return None


# ---------------------------------------------------------------- main

def main():
    print("discovering front-month contracts...", flush=True)
    mapping = discover_products()
    print("tracking:", mapping, flush=True)
    product_ids = list(mapping.values())
    if not product_ids:
        raise SystemExit("no products discovered; check REST connectivity")

    db = open_db()
    threading.Thread(target=poll_books, args=(db, product_ids), daemon=True).start()
    threading.Thread(target=poll_meta, args=(db, product_ids), daemon=True).start()

    print(f"recording to {DB_PATH} — leave running through 2-3 expiry cycles",
          flush=True)
    try:
        while True:
            time.sleep(3600)
    except KeyboardInterrupt:
        print("stopped")


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Live paper forward test for the Coinbase stat-arb v3 (hill-climb improved).

Runs every 15 min via cron. Each pass:
  1. Loads the latest market state — executable L2 from the recorder DB when
     fresh (<5 min), else REST mids with the spread haircut.
  2. Computes net carry C per pair, applies the 60-second L2 persistence gate
     (or 3-consecutive-run gate on REST fallback).
  3. Enters/exits PAPER positions under the improved policy
     (entry 2.0%, standdown 1.5%, session gates, 3x margin guard).
  4. Accrues funding pro-rata, marks to market, writes forward_status.json.

Paper/research only. No orders, no API keys, no account access.
"""

import copy
import json
import os
import sqlite3
import sys
import time

import requests

sys.path.insert(0, "/home/hatch/workspace/stat-arb/code")
from engine import (Bar, Ledger, Position, decide, mtm_equity, net_carry_pct,
                    session_mult, contract_expiry_ts)  # noqa: E402
from pareto_hillclimb import CONTRACTS, FUNDING_NOW  # noqa: E402

REST = "https://api.coinbase.com/api/v3/brokerage"
EXPIRY = {pair: contract_expiry_ts(dated) for pair, (dated, _) in CONTRACTS.items()}
CODE = "/home/hatch/workspace/stat-arb/code"
RECORDER_DB = os.path.join(CODE, "stat_arb_data.db")
PAPER_DB = os.path.join(CODE, "paper_state.db")
STATUS_JSON = os.path.join(CODE, "forward_status.json")

# Hill-climb improved policy, locked 2026-10-05
# (in-sample 14.85% APY on 12.4d real hourly data vs 10.87% at v3 defaults)
POLICY = {
    "entry_C": 2.0,
    "standdown_C": 1.0,
    "persist_runs": 2,          # consecutive 15-min runs (REST fallback)
    "persist_polls": 6,         # consecutive 10s book polls = 60s (recorder)
    "clip_usd": {"BTC": 35000.0, "XRP": 10000.0},
    "use_sessions": True,
    "spread_haircut_bps": 8.0,  # REST-mid fallback only
}
EQUITY = 50000.0


# ---------------------------------------------------------------- state

def funding_days_logged(db_path=RECORDER_DB, perp_id="XPP-20DEC30-CDE"):
    """Distinct UTC days of hourly funding logged for the XRP perp (judge gate)."""
    if not os.path.exists(db_path):
        return 0
    try:
        db = sqlite3.connect(db_path)
        n = db.execute(
            """SELECT COUNT(DISTINCT date(ts,'unixepoch')) FROM funding
               WHERE product_id=?""", (perp_id,)).fetchone()[0]
        db.close()
        return n or 0
    except Exception:
        return 0


def paper_db():
    db = sqlite3.connect(PAPER_DB)
    db.execute("CREATE TABLE IF NOT EXISTS kv(key TEXT PRIMARY KEY, value TEXT)")
    db.execute("""CREATE TABLE IF NOT EXISTS actions(
        ts REAL, pair TEXT, action TEXT, detail TEXT)""")
    return db


def kv_get(db, key, default=None):
    r = db.execute("SELECT value FROM kv WHERE key=?", (key,)).fetchone()
    return json.loads(r[0]) if r else default


def kv_put(db, key, value):
    db.execute("INSERT OR REPLACE INTO kv VALUES (?,?)", (key, json.dumps(value)))
    db.commit()


def load_ledger(db):
    st = kv_get(db, "ledger") or {}
    led = Ledger(equity=EQUITY)
    led.cash_pnl = st.get("cash_pnl", 0.0)
    led.funding_pnl = st.get("funding_pnl", 0.0)
    led.fees_paid = st.get("fees_paid", 0.0)
    for pair, p in st.get("positions", {}).items():
        led.positions[pair] = Position(**p)
    return led


def save_ledger(db, led):
    kv_put(db, "ledger", {
        "cash_pnl": led.cash_pnl,
        "funding_pnl": led.funding_pnl,
        "fees_paid": led.fees_paid,
        "positions": {k: vars(v) for k, v in led.positions.items()},
    })


# ---------------------------------------------------------------- market state

def rest_state():
    """Latest mids + funding via REST. Returns {pair: Bar(executable=False)}."""
    out = {}
    for pair, (dated_id, perp_id) in CONTRACTS.items():
        dated = requests.get(f"{REST}/market/products/{dated_id}",
                             timeout=20).json()
        perp = requests.get(f"{REST}/market/products/{perp_id}",
                            timeout=20).json()
        dp = dated.get("product", dated)
        pp = perp.get("product", perp)
        fpd = pp.get("future_product_details", {}) or {}
        fr = fpd.get("funding_rate")
        out[pair] = Bar(
            ts=time.time(),
            dated=float(dp["price"]), perp=float(pp["price"]),
            funding_hr=float(fr) if fr not in (None, "") else FUNDING_NOW[pair],
            expiry_ts=EXPIRY[pair], executable=False)
    return out


def recorder_state():
    """
    Latest executable state from the recorder DB (10s REST book polls).
    Returns ({pair: Bar(executable=True)}, {pair: consecutive polls holding})
    or (None, None) when the recorder is stale/missing.
    """
    if not os.path.exists(RECORDER_DB):
        return None, None
    db = sqlite3.connect(RECORDER_DB)
    now = time.time()
    bars, gates = {}, {}
    for pair, (dated_id, perp_id) in CONTRACTS.items():
        r = db.execute(
            "SELECT ts,bid_px,ask_px FROM books WHERE product_id=? ORDER BY ts DESC LIMIT 1",
            (dated_id,)).fetchone()
        q = db.execute(
            "SELECT ts,bid_px,ask_px FROM books WHERE product_id=? ORDER BY ts DESC LIMIT 1",
            (perp_id,)).fetchone()
        if not r or not q or now - r[0] > 120 or now - q[0] > 120:
            return None, None
        fr = db.execute(
            "SELECT rate FROM funding WHERE product_id=? ORDER BY ts DESC LIMIT 1",
            (perp_id,)).fetchone()
        # 60-second persistence = 6 consecutive 10s polls holding the gate
        snaps = db.execute(
            """SELECT b1.ts, b1.bid_px, b2.ask_px FROM books b1
               JOIN books b2 ON ABS(b1.ts-b2.ts)<6
               WHERE b1.product_id=? AND b2.product_id=? AND b1.ts>? ORDER BY b1.ts""",
            (dated_id, perp_id, now - 120)).fetchall()
        held = 0
        for ts, dbid, pask in sorted(snaps):
            b = Bar(ts=ts, dated=dbid, perp=pask,
                    funding_hr=float(fr[0]) if fr else FUNDING_NOW[pair],
                    expiry_ts=EXPIRY[pair], executable=True)
            if net_carry_pct(b, 0.0) >= POLICY["entry_C"]:
                held += 1
            else:
                held = 0
        bars[pair] = Bar(ts=now, dated=r[1], perp=q[2],
                         funding_hr=float(fr[0]) if fr else FUNDING_NOW[pair],
                         expiry_ts=EXPIRY[pair], executable=True)
        gates[pair] = held  # consecutive ~10s polls holding the gate
    return bars, gates


# ---------------------------------------------------------------- main pass

def main():
    db = paper_db()
    led = load_ledger(db)
    prev_runs = kv_get(db, "run_gates") or {}
    started = kv_get(db, "started_utc")

    bars, gates = recorder_state()
    source = "recorder_l2"
    if bars is None:
        bars = rest_state()
        gates = None
        source = "rest_mid_fallback"

    params = {
        "entry_C": POLICY["entry_C"],
        "standdown_C": POLICY["standdown_C"],
        "persist_bars": 1,
        "clip_usd": POLICY["clip_usd"],
        "use_sessions": POLICY["use_sessions"],
        "spread_haircut_bps": 0.0 if source == "recorder_l2"
        else POLICY["spread_haircut_bps"],
    }

    now = time.time()
    actions_taken = []
    new_run_gates = {}
    for pair, bar in bars.items():
        C = net_carry_pct(bar, params["spread_haircut_bps"])
        in_pos = pair in led.positions
        if not in_pos:
            if gates is not None:
                gate_ok = gates[pair] >= POLICY["persist_polls"]
            else:
                streak = prev_runs.get(pair, 0)
                streak = streak + 1 if C >= POLICY["entry_C"] else 0
                new_run_gates[pair] = streak
                gate_ok = streak >= POLICY["persist_runs"]
            # margin guard: >=3x intraday maintenance on entry
            if gate_ok:
                clip = POLICY["clip_usd"][pair] * (
                    session_mult(now) if POLICY["use_sessions"] else 1.0)
                maint = clip * 2 * 0.10  # ~10% intraday per leg
                free = EQUITY + led.cash_pnl - sum(
                    p.notional * 2 * 0.10 for p in led.positions.values())
                if clip > 0 and free >= 3 * maint:
                    window = {pair: [bar]}
                    for a in decide(window, params, led):
                        a["source"] = source
                        actions_taken.append(a)
                        db.execute("INSERT INTO actions VALUES (?,?,?,?)",
                                   (now, pair, a["action"], json.dumps(a)))
                elif clip > 0:
                    actions_taken.append(
                        {"pair": pair, "action": "SKIP",
                         "reason": "margin guard: <3x maintenance"})
        else:
            # exits + accrual need the position's own bar window
            window = {pair: [bar]}
            for a in decide(window, params, led):
                a["source"] = source
                actions_taken.append(a)
                db.execute("INSERT INTO actions VALUES (?,?,?,?)",
                           (now, pair, a["action"], json.dumps(a)))

    # funding accrual pro-rata since last run (15 min = 0.25h)
    last = kv_get(db, "last_ts") or now
    elapsed_h = min((now - last) / 3600.0, 1.0)
    for pair, pos in led.positions.items():
        led.funding_pnl += -bars[pair].funding_hr * pos.notional * elapsed_h
    kv_put(db, "last_ts", now)
    if gates is None:
        kv_put(db, "run_gates", new_run_gates)
    if not started:
        kv_put(db, "started_utc",
               time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)))
        started = kv_get(db, "started_utc")
    save_ledger(db, led)
    db.commit()

    eq = mtm_equity({pair: [bar] for pair, bar in bars.items()}, led)
    xrp_days = funding_days_logged()
    status = {
        "updated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(now)),
        "started_utc": started,
        "source": source,
        "policy": POLICY,
        "policy_notes": {
            # Judge gate (Claude must-fix 4): XRP not sized until 7 days of
            # realized XPP funding are logged. Paper entries before that are
            # data collection, not validated signals.
            "xrp_funding_days_logged": xrp_days,
            "xrp_validated": xrp_days >= 7,
            "eth_sleeve": "removed (fee floor uneconomic; judge must-fix)",
            "notional_cap": "0.9x equity (judge must-fix)",
        },
        "equity_mtm": round(eq, 2),
        "total_pnl": round(eq - EQUITY, 2),
        "cash_pnl": round(led.cash_pnl, 2),
        "funding_pnl": round(led.funding_pnl, 2),
        "fees_paid": round(led.fees_paid, 2),
        "positions": {
            pair: {
                "notional": p.notional,
                "entry_basis_bps": round(p.entry_basis_bps, 1),
                "net_carry_now": round(
                    net_carry_pct(bars[pair], params["spread_haircut_bps"]), 2),
            } for pair, p in led.positions.items()},
        "signals": {
            pair: {
                "net_carry_pct": round(
                    net_carry_pct(bar, params["spread_haircut_bps"]), 2),
                "in_position": pair in led.positions,
            } for pair, bar in bars.items()},
        "actions_this_run": actions_taken,
    }
    with open(STATUS_JSON, "w") as f:
        json.dump(status, f, indent=1)
    n_act = db.execute("SELECT COUNT(*) FROM actions").fetchone()[0]
    print(json.dumps({"equity_mtm": status["equity_mtm"],
                      "positions": list(status["positions"]),
                      "actions": actions_taken,
                      "lifetime_actions": n_act,
                      "source": source}))
    db.close()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
"""
Pareto hill-climbing optimizer for the Coinbase stat-arb v3 parameters.

Optimizes (entry_C, standdown_C, persist_bars, clips, session filter, haircut)
over three objectives: maximize net APY, maximize Sharpe, minimize max drawdown.
Keeps the full Pareto frontier — no single weighted score hides trade-offs.

Data: hourly candle mids for the live front-month contracts (real, but short:
~12 days as of 2026-10-05; expired contracts are delisted so no stitching).
Funding history has no public endpoint -> current observed hourly funding is
held constant over the window (STATED ASSUMPTION, revisited by the recorder).

Usage:
  python3 pareto_hillclimb.py            # run on candle history, print frontier
  python3 pareto_hillclimb.py --db stat_arb_data.db   # run on recorder data (>=7d)

Paper/research only.
"""

import copy
import itertools
import json
import math
import random
import sqlite3
import sys
import time

import requests

sys.path.insert(0, "/home/hatch/workspace/stat-arb/code")
from engine import V3_DEFAULTS, Bar, backtest, contract_expiry_ts  # noqa: E402

REST = "https://api.coinbase.com/api/v3/brokerage"
CONTRACTS = {
    "BTC": ("BIT-30OCT26-CDE", "BIP-20DEC30-CDE"),
    "XRP": ("XRP-30OCT26-CDE", "XPP-20DEC30-CDE"),
}
# Dated-leg true expiry from contract metadata; perps are 2089-dated (use dated leg).
EXPIRY = {pair: contract_expiry_ts(dated) for pair, (dated, _) in CONTRACTS.items()}
FUNDING_NOW = {"BTC": 0.000003, "XRP": -0.000002}  # observed 2026-10-05, per hour


# ---------------------------------------------------------------- data

def load_candle_history():
    """Return {pair: [Bar]} from hourly candles. Mids => executable=False."""
    out = {}
    for pair, (dated_id, perp_id) in CONTRACTS.items():
        series = {}
        for pid, key in ((dated_id, "dated"), (perp_id, "perp")):
            r = requests.get(f"{REST}/market/products/{pid}/candles",
                             params={"granularity": "ONE_HOUR", "limit": 300},
                             timeout=30)
            r.raise_for_status()
            for c in r.json().get("candles", []):
                ts = int(c["start"])
                mid = (float(c["open"]) + float(c["close"])) / 2
                series.setdefault(ts, {})[key] = mid
        bars = [Bar(ts=t, dated=v["dated"], perp=v["perp"],
                    funding_hr=FUNDING_NOW[pair], expiry_ts=EXPIRY[pair])
                for t, v in sorted(series.items())
                if "dated" in v and "perp" in v]
        out[pair] = bars
        print(f"  {pair}: {len(bars)} hourly bars, "
              f"{time.strftime('%m-%d %H:%M', time.gmtime(bars[0].ts))} -> "
              f"{time.strftime('%m-%d %H:%M', time.gmtime(bars[-1].ts))} UTC",
              flush=True)
    return out


def load_recorder_db(path, min_days=7):
    """Build Bar series from recorder L2 snapshots (executable bid/ask)."""
    db = sqlite3.connect(path)
    # map product ids back to pairs via contracts table-less heuristic
    cur = db.execute("SELECT DISTINCT product_id FROM books")
    pids = [r[0] for r in cur.fetchall()]
    if not pids:
        raise SystemExit("recorder DB has no book data yet")
    # resolve dated/perp per root by expiry embedded in product id
    pairs = {}
    for pair, (d, p_) in CONTRACTS.items():
        pairs[pair] = (d, p_)
    out = {pair: [] for pair in pairs}
    for pair, (dated_id, perp_id) in pairs.items():
        rows = db.execute(
            """SELECT ts, bid_px, ask_px FROM books WHERE product_id=?
               ORDER BY ts""", (dated_id,)).fetchall()
        perps = {int(r[0] / 3600): r for r in db.execute(
            "SELECT ts, bid_px, ask_px FROM books WHERE product_id=? ORDER BY ts",
            (perp_id,)).fetchall()}
        fund = db.execute(
            "SELECT rate FROM funding WHERE product_id=? ORDER BY ts DESC LIMIT 1",
            (perp_id,)).fetchone()
        fr = float(fund[0]) if fund else FUNDING_NOW[pair]
        for ts, bb, ba in rows:
            key = int(ts / 3600)
            if key not in perps:
                continue
            _, pb, pa = perps[key]
            out[pair].append(Bar(ts=ts, dated=bb, perp=pa, funding_hr=fr,
                                 expiry_ts=EXPIRY[pair], executable=True))
    span_days = 0
    for pair, bars in out.items():
        if len(bars) > 1:
            span_days = max(span_days, (bars[-1].ts - bars[0].ts) / 86400)
    print(f"recorder data: {span_days:.1f} days", flush=True)
    if span_days < min_days:
        raise SystemExit(f"only {span_days:.1f}d of recorder data (< {min_days}d); "
                         "run again when the forward test has accumulated more")
    return out


# ---------------------------------------------------------------- evaluation

def evaluate(params, data):
    metrics, _ = backtest(
        {p: [Bar(b.ts, b.dated, b.perp, b.funding_hr, b.expiry_ts, b.executable)
             for b in bars] for p, bars in data.items()},
        FUNDING_NOW, params)
    return metrics


def dominates(a, b):
    """a dominates b on (apy, sharpe, -maxdd), requiring >=1 trade."""
    if a["n_trades"] < 1:
        return False
    if b["n_trades"] < 1:
        return True
    better = False
    for key, sign in (("apy_pct", 1), ("sharpe", 1), ("maxdd_pct", -1)):
        va, vb = sign * a[key], sign * b[key]
        if va < vb - 1e-9:
            return False
        if va > vb + 1e-9:
            better = True
    return better


# ---------------------------------------------------------------- search

def neighbors(params):
    """One-step perturbations of the parameter vector."""
    cands = []
    p = copy.deepcopy(params)

    def var(key, deltas):
        for d in deltas:
            q = copy.deepcopy(p)
            q[key] = round(p[key] + d, 2)
            if key == "entry_C" and q[key] <= q["standdown_C"]:
                continue
            if key == "standdown_C" and q[key] >= q["entry_C"]:
                continue
            cands.append(q)

    var("entry_C", [-1.0, -0.5, 0.5, 1.0])
    var("standdown_C", [-0.5, 0.5, 1.0])
    var("spread_haircut_bps", [-4.0, 4.0, 8.0])
    for d in (-1, 1):
        q = copy.deepcopy(p)
        q["persist_bars"] = max(1, p["persist_bars"] + d)
        if q["persist_bars"] <= 4:
            cands.append(q)
    for pair in ("BTC", "XRP"):
        for mult in (0.5, 2.0):
            q = copy.deepcopy(p)
            q["clip_usd"] = dict(p["clip_usd"])
            q["clip_usd"][pair] = round(p["clip_usd"][pair] * mult, 0)
            cands.append(q)
    q = copy.deepcopy(p)
    q["use_sessions"] = not p["use_sessions"]
    cands.append(q)
    return cands


def hill_climb(data, seeds, max_iters=40):
    archive = []  # list of (params, metrics), mutually non-dominated
    seen = set()

    def key_of(params):
        return json.dumps(params, sort_keys=True)

    def consider(params):
        k = key_of(params)
        if k in seen:
            return None
        seen.add(k)
        m = evaluate(params, data)
        # prune archive
        nonlocal_archive = [e for e in archive if not dominates(m, e[1])]
        if any(dominates(e[1], m) for e in nonlocal_archive):
            return None
        nonlocal_archive.append((copy.deepcopy(params), m))
        archive[:] = nonlocal_archive
        return m

    for s, seed in enumerate(seeds):
        print(f"--- seed {s + 1}/{len(seeds)}: entry_C={seed['entry_C']} "
              f"standdown={seed['standdown_C']} persist={seed['persist_bars']}",
              flush=True)
        cur = copy.deepcopy(seed)
        cur_m = consider(cur)
        if cur_m is None:
            # seed dominated by the archive already: still walk from it,
            # just don't force it into the archive
            cur_m = evaluate(cur, data)
        for it in range(max_iters):
            improved = False
            random.shuffle(ns := neighbors(cur))
            for nb in ns:
                m = consider(nb)
                if m is None:
                    continue
                if cur_m is None or dominates(m, cur_m):
                    cur, cur_m = nb, m
                    improved = True
                    break
            if not improved:
                break
        print(f"    settled at apy={cur_m['apy_pct']}% sharpe={cur_m['sharpe']} "
              f"dd={cur_m['maxdd_pct']}% trades={cur_m['n_trades']}", flush=True)
    return archive


def main():
    use_db = None
    if "--db" in sys.argv:
        use_db = sys.argv[sys.argv.index("--db") + 1]
    print("loading data...", flush=True)
    data = load_recorder_db(use_db) if use_db else load_candle_history()

    seeds = []
    base = copy.deepcopy(V3_DEFAULTS)
    seeds.append(base)
    for ec, sc in [(2.0, 1.0), (4.0, 2.0), (1.5, 0.75), (3.0, 2.0)]:
        s = copy.deepcopy(base)
        s["entry_C"], s["standdown_C"] = ec, sc
        seeds.append(s)

    t0 = time.time()
    archive = hill_climb(data, seeds)
    print(f"\nsearch done in {time.time() - t0:.0f}s, "
          f"frontier size {len(archive)}", flush=True)

    frontier = sorted(
        ({"params": p, "metrics": m} for p, m in archive),
        key=lambda e: -e["metrics"]["apy_pct"])
    with open("/home/hatch/workspace/stat-arb/code/frontier.json", "w") as f:
        json.dump({"generated_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
                   "data_source": use_db or "coinbase_hourly_candles",
                   "n_bars": {p: len(b) for p, b in data.items()},
                   "frontier": frontier}, f, indent=1)

    print("\n=== PARETO FRONTIER (apy% / sharpe / maxdd% / trades) ===")
    for e in frontier:
        p, m = e["params"], e["metrics"]
        print(f"  apy {m['apy_pct']:>7.2f}%  sharpe {m['sharpe']:>6.2f}  "
              f"dd {m['maxdd_pct']:>5.2f}%  n {m['n_trades']:>2}  "
              f"entry {p['entry_C']}% / stand {p['standdown_C']}% / "
              f"persist {p['persist_bars']} / haircut {p['spread_haircut_bps']}bps / "
              f"sess {p['use_sessions']} / clip {p['clip_usd']}")

    # sensitivity of the top-apy point to the executable haircut
    if frontier:
        top = frontier[0]["params"]
        print("\n=== haircut sensitivity (top-apy params) ===")
        for h in (0.0, 8.0, 16.0):
            q = copy.deepcopy(top)
            q["spread_haircut_bps"] = h
            m = evaluate(q, data)
            print(f"  haircut {h:>4.0f} bps -> apy {m['apy_pct']:>7.2f}%  "
                  f"trades {m['n_trades']}")


if __name__ == "__main__":
    main()