Repository Public
1"""Strategy loader: identity, META validation, purity guard, weight validation."""2from __future__ import annotations34import builtins5import hashlib6import io7import os8import socket9import time1011import numpy as np12import pytest1314from conftest import make_data, write_strategy15from spq.strategy import (HashMismatch, PurityError, StrategyError, WeightsError, load_strategy, purity_guard,16 sha256_file, static_check, validate_meta, validate_weights)1718GOOD_META = {"name": "eth_trend", "version": "1", "universe": ["ETH", "LINK"], "interval": "1h", "lookback": 720,19 "params": {"fast": 24, "slow": 168}}202122def test_identity_is_the_sha256_of_the_file(strategy_dir):23 d = strategy_dir()24 s = load_strategy(d)25 blob = (d / "strategy.py").read_bytes()26 assert s.sha256 == hashlib.sha256(blob).hexdigest() == sha256_file(d / "strategy.py")27 assert load_strategy(d, s.sha256).sha256 == s.sha25628 assert load_strategy(d, s.sha256.upper()).sha256 == s.sha25629 (d / "strategy.py").write_bytes(blob + b"\n# one more byte\n")30 assert load_strategy(d).sha256 != s.sha25631 with pytest.raises(HashMismatch):32 load_strategy(d, s.sha256)33 assert not list(d.glob("__pycache__")) # loading leaves nothing behind343536def test_hash_mismatch_is_raised_before_any_strategy_code_runs(tmp_path):37 d = write_strategy(tmp_path / "boom", "", raw="raise RuntimeError('module code executed')\n")38 with pytest.raises(HashMismatch):39 load_strategy(d, "0" * 64)40 with pytest.raises(StrategyError, match="failed at import"):41 load_strategy(d)424344def test_loader_accepts_the_architecture_example(tmp_path):45 raw = (f"META = {GOOD_META!r}\n"46 "def target_weights(data, params):\n"47 " return {s: 0.5 for s in data}\n")48 s = load_strategy(write_strategy(tmp_path / "eth_trend" / "1", "", raw=raw))49 assert s.name == "eth_trend" and s.universe == ["ETH", "LINK"] and s.lookback == 72050 assert s.params == {"fast": 24, "slow": 168}51 assert s(make_data(30, ("ETH", "LINK"))) == {"ETH": 0.5, "LINK": 0.5}52 assert load_strategy(tmp_path / "eth_trend" / "1" / "strategy.py").sha256 == s.sha256535455@pytest.mark.parametrize("change", [56 {"name": ""}, {"version": 1}, {"universe": []}, {"universe": ["eth"]}, {"universe": ["ETH", "ETH"]},57 {"universe": "ETH"}, {"interval": "1x"}, {"lookback": 0}, {"lookback": 2.5}, {"lookback": True},58 {"params": [1, 2]}, {"params": {"x": float("nan")}}, {"params": {"f": print}}, {"extra": 1},59])60def test_bad_meta_is_refused(change):61 with pytest.raises(StrategyError):62 validate_meta({**GOOD_META, **change})636465def test_missing_meta_key_and_missing_pieces(tmp_path, strategy_dir):66 for key in GOOD_META:67 with pytest.raises(StrategyError, match="missing"):68 validate_meta({k: v for k, v in GOOD_META.items() if k != key})69 with pytest.raises(StrategyError, match="no strategy.py"):70 load_strategy(tmp_path / "nowhere")71 with pytest.raises(StrategyError, match="does not define META"):72 load_strategy(write_strategy(tmp_path / "a", "", raw="def target_weights(data, params):\n return {}\n"))73 with pytest.raises(StrategyError, match="target_weights"):74 load_strategy(write_strategy(tmp_path / "b", "", raw=f"META = {GOOD_META!r}\n"))75 with pytest.raises(StrategyError, match="syntax error"):76 load_strategy(write_strategy(tmp_path / "c", "", raw="def broken(:\n"))777879@pytest.mark.parametrize("line,word", [80 ("import os", "os"), ("import socket", "socket"), ("import requests", "requests"), ("import time", "time"),81 ("from datetime import datetime", "datetime"), ("import random", "random"), ("import urllib.request", "urllib"),82 ("from pathlib import Path", "pathlib"), ("import subprocess", "subprocess"), ("from . import helper", "None"),83 ("x = open('/etc/hosts').read()", "open"), ("x = eval('1')", "eval"), ("x = __import__('os')", "__import__"),84 ("import pandas as pd\nx = pd.Timestamp.now()", ".now"), ("import pandas as pd\nx = pd.read_csv('f')", "read_csv"),85 ("import numpy as np\nx = np.load('f')", ".load"), ("x = ().__class__.__subclasses__()", "__subclasses__"),86 ("COUNTER = 0\ndef bump():\n global COUNTER\n COUNTER += 1", "global"),87])88def test_static_check_refuses_impure_sources(tmp_path, line, word):89 problems = static_check(line)90 assert problems and word in " ".join(problems)91 raw = line + f"\nMETA = {GOOD_META!r}\ndef target_weights(data, params):\n return {{}}\n"92 d = write_strategy(tmp_path / "bad", "", raw=raw)93 with pytest.raises(StrategyError, match="purity check"):94 load_strategy(d)959697def test_static_check_accepts_numerical_code():98 src = ("import math\nimport numpy as np\nimport pandas as pd\nfrom scipy import stats\nfrom arch import arch_model\n"99 "from statsmodels.tsa.stattools import adfuller\nfrom sklearn.linear_model import LinearRegression\n"100 "def f(df):\n return float(np.log(df['close']).diff().rolling(5).std().iloc[-1])\n")101 assert static_check(src) == []102103104def test_runtime_guard_blocks_network_files_clock_and_randomness():105 real_open, real_time = builtins.open, time.time106 with purity_guard():107 for call in (lambda: open(__file__), lambda: io.open(__file__), lambda: os.open(__file__, os.O_RDONLY),108 lambda: socket.socket(), lambda: socket.create_connection(("127.0.0.1", 9)),109 lambda: socket.getaddrinfo("example.com", 80), lambda: time.time(), lambda: time.time_ns(),110 lambda: time.sleep(0), lambda: os.urandom(4), lambda: os.system("true"),111 lambda: np.random.rand(2), lambda: np.random.default_rng(), lambda: np.random.normal()):112 with pytest.raises(PurityError):113 call()114 with purity_guard(): # re-entrant115 with pytest.raises(PurityError):116 time.time()117 with pytest.raises(PurityError): # still active after the inner block118 time.time()119 assert float(np.arange(5).mean()) == 2.0 # numerical code is unaffected120 assert builtins.open is real_open and time.time is real_time121 assert isinstance(time.time(), float) and open(__file__).close() is None122123124def test_guard_is_applied_to_loaded_strategies_and_restored_on_error(strategy_dir):125 # passes the static check (no forbidden name) but draws random numbers at run time126 d = strategy_dir("import numpy as np\nreturn {s: float(np.random.rand()) / 2 for s in data}\n")127 s = load_strategy(d)128 with pytest.raises(PurityError, match="random"):129 s(make_data(20))130 assert np.random.rand() < 1.0 # entry points are restored after the failure131 # strict=False lets a researcher run an unchecked file; the runtime guard still applies132 d = strategy_dir("import time\nreturn {s: time.time() % 1 / 2 for s in data}\n")133 loose = load_strategy(d, strict=False)134 assert loose.static_problems135 with pytest.raises(PurityError, match="clock"):136 loose(make_data(20))137138139def test_validate_weights():140 u = ["A", "B", "C"]141 assert validate_weights({"A": 0.5, "B": 0.25}, u) == {"A": 0.5, "B": 0.25, "C": 0.0}142 assert validate_weights({}, u) == {"A": 0.0, "B": 0.0, "C": 0.0}143 assert validate_weights({"A": np.float64(1.0), "B": 0, "C": np.int64(0)}, u) == {"A": 1.0, "B": 0.0, "C": 0.0}144 snapped = validate_weights({"A": 1 / 3 + 3e-10, "B": 1 / 3, "C": 1 / 3}, u) # rounding noise only145 assert sum(snapped.values()) <= 1.0 and snapped["A"] == pytest.approx(1 / 3)146 assert validate_weights({"A": -1e-12, "B": 1.0 + 1e-12}, u) == {"A": 0.0, "B": 1.0, "C": 0.0}147 for bad, msg in [148 ({"A": 1.01}, "outside"), ({"A": -0.01}, "outside"), ({"A": 0.6, "B": 0.6}, "sum"),149 ({"A": float("nan")}, "finite"), ({"A": float("inf")}, "finite"), ({"A": "0.5"}, "not a number"),150 ({"A": None}, "not a number"), ({"A": True}, "not a number"), ({"Z": 0.1}, "outside the universe"),151 ([0.5, 0.5], "dict"), (None, "dict"), ({"A": 0.34, "B": 0.34, "C": 0.34}, "sum"),152 ]:153 with pytest.raises(WeightsError, match=msg):154 validate_weights(bad, u)155