spb/sp-wallet

1from __future__ import annotations23import textwrap4from pathlib import Path56import numpy as np7import pandas as pd8import pytest910ROOT = Path(__file__).resolve().parents[1]11# v7: directories named rj_<job> are strategies installed by research jobs on a deployed node: not the library12_ALL = sorted(p.parent for p in (ROOT / "strategies").glob("*/*/strategy.py") if not p.parent.parent.name.startswith("rj_"))13# Perpetual-futures strategies (contract v2, directories named <name>_perp) have their own tests14# (test_perp_library.py); LIBRARY stays the spot library that the version-1 tests were written for.15PERP_LIBRARY = [d for d in _ALL if d.parent.name.endswith("_perp")]16LIBRARY = [d for d in _ALL if d not in PERP_LIBRARY]171819def pytest_collection_modifyitems(config, items):20    import os2122    if os.environ.get("SPQ_NETWORK_TESTS") == "1":23        return24    skip = pytest.mark.skip(reason="live network test: set SPQ_NETWORK_TESTS=1")25    for item in items:26        if "network" in item.keywords:27            item.add_marker(skip)282930def make_series(n: int, seed: int = 0, start: str = "2024-01-01", freq: str = "1h", sigma: float = 0.01,31                drift: float = 0.0, p0: float = 100.0) -> pd.DataFrame:32    """Geometric random walk OHLCV with i.i.d. returns; open[t] == close[t-1]."""33    rng = np.random.default_rng(seed)34    r = rng.normal(drift, sigma, n)35    close = p0 * np.exp(np.cumsum(r))36    open_ = np.concatenate([[p0], close[:-1]])37    wick = np.abs(rng.normal(0, sigma / 2, (2, n)))38    high = np.maximum(open_, close) * (1 + wick[0])39    low = np.minimum(open_, close) * (1 - wick[1])40    idx = pd.date_range(start, periods=n, freq=freq, tz="UTC", name="ts")41    return pd.DataFrame({"open": open_, "high": high, "low": low, "close": close,42                         "volume": rng.uniform(1, 10, n)}, index=idx)434445def make_data(n: int, symbols=("AAA", "BBB"), seed: int = 0, **kw) -> dict[str, pd.DataFrame]:46    return {s: make_series(n, seed + 17 * i, **kw) for i, s in enumerate(symbols)}474849def frame(opens, closes, start="2024-01-01", freq="1h") -> pd.DataFrame:50    """Hand-written bars (high/low derived)."""51    o, c = np.asarray(opens, float), np.asarray(closes, float)52    idx = pd.date_range(start, periods=len(o), freq=freq, tz="UTC", name="ts")53    return pd.DataFrame({"open": o, "high": np.maximum(o, c), "low": np.minimum(o, c), "close": c,54                         "volume": np.ones(len(o))}, index=idx)555657def write_strategy(directory: Path, body: str, meta: dict | None = None, raw: str | None = None) -> Path:58    """Create <directory>/strategy.py. `body` is the body of target_weights(data, params)."""59    directory.mkdir(parents=True, exist_ok=True)60    if raw is None:61        meta = meta or {"name": "t", "version": "1", "universe": ["AAA", "BBB"], "interval": "1h", "lookback": 10,62                        "params": {}}63        raw = f"META = {meta!r}\n\n\ndef target_weights(data, params):\n" + textwrap.indent(textwrap.dedent(body), "    ")64    (directory / "strategy.py").write_text(raw)65    return directory666768@pytest.fixture69def strategy_dir(tmp_path):70    def make(body: str = "return {s: 0.5 for s in data}\n", **kw) -> Path:71        return write_strategy(tmp_path / f"s{len(list(tmp_path.iterdir()))}", body, **kw)72    return make73