Repository Public
1"""Helpers shared by the perp tests (hand-written bars, scripted decisions, synthetic funding)."""2from __future__ import annotations34from decimal import Decimal56import numpy as np7import pandas as pd89from conftest import make_data10from spq.perp_engine import PerpConfig1112D = Decimal131415def bars(opens, closes, highs=None, lows=None, funding=None, start="2024-01-01", freq="1h", **extra) -> pd.DataFrame:16 """Hand-written bars. High/low default to the open/close envelope; `funding[i]` is the rate settled at ts[i]."""17 o, c = np.asarray(opens, float), np.asarray(closes, float)18 h = np.maximum(o, c) if highs is None else np.asarray(highs, float)19 low = np.minimum(o, c) if lows is None else np.asarray(lows, float)20 idx = pd.date_range(start, periods=len(o), freq=freq, tz="UTC", name="ts")21 df = pd.DataFrame({"open": o, "high": h, "low": low, "close": c, "volume": np.ones(len(o))}, index=idx)22 if funding is not None:23 df["funding"] = np.asarray(funding, float)24 for k, v in extra.items():25 df[k] = np.asarray(v, float)26 return df272829def script(*steps):30 """A decision function that returns the given weights one call after the other (None = leave as is)."""31 calls = iter(steps)3233 def decide(_view):34 return next(calls, None)3536 return decide373839def scripted_strategy(*steps, universe=("X",), max_leverage=20.0):40 """A perp Strategy (weights validated against META) that plays the given weights one call after the other;41 once they are exhausted it keeps returning the last one."""42 from spq.strategy import from_function4344 state = {"i": 0}4546 def f(_data, _params):47 w = steps[min(state["i"], len(steps) - 1)]48 state["i"] += 149 return w5051 return from_function(f, perp_meta(universe, lookback=1, max_leverage=max_leverage))525354def cfg(**kw) -> PerpConfig:55 """Round numbers for hand computation: fee 10 bps, slippage 10 bps, 10x, maintenance 5 %, no penalty."""56 base = dict(interval="1h", taker_fee_bps=10.0, slippage_bps=10.0, initial_cash=1000.0, min_order=0.0,57 leverage=10.0, maintenance_fraction=0.05, liquidation_penalty_bps=0.0)58 base.update(kw)59 return PerpConfig(**base)606162def perp_meta(universe=("X",), lookback=10, max_leverage=3.0, max_gross=None, params=None, name="t"):63 return {"name": name, "version": "1", "kind": "perp", "universe": list(universe), "interval": "1h",64 "lookback": lookback, "max_leverage": max_leverage,65 "max_gross_exposure": max_gross if max_gross is not None else max_leverage * len(universe),66 "params": params or {}}676869def perp_data(n, symbols=("BTC", "ETH"), seed=0, rate=1.25e-5, rate_sigma=2e-5, **kw) -> dict[str, pd.DataFrame]:70 """Random-walk candles with an autocorrelated funding column (settled at each bar's open)."""71 data = make_data(n, symbols, seed=seed, **kw)72 rng = np.random.default_rng(seed + 999)73 for i, s in enumerate(symbols):74 noise = rng.normal(0, rate_sigma, n)75 f = np.empty(n)76 f[0] = rate77 for t in range(1, n):78 f[t] = rate + 0.9 * (f[t - 1] - rate) + 0.3 * noise[t]79 data[s]["funding"] = np.round(f, 10)80 data[s]["premium"] = np.round(f * 8 - 1e-4, 10)81 return data828384def settlements(df: pd.DataFrame) -> pd.DataFrame:85 """The settlement series that `df['funding']` was built from (index = settlement hour)."""86 f = df[["funding"]].rename(columns={"funding": "rate"}).dropna()87 f["premium"] = df["premium"] if "premium" in df.columns else np.nan88 f["time_ms"] = f.index.as_unit("ms").asi8 + 4089 return f90