ftdt-quant-lab/tests/test_system.py

#!/usr/bin/env python3
"""Tests for FTDT Quant Lab — signal generation, backtest, and API validation.
Run: .venv/bin/python tests/test_system.py  (requires venv)"""
import sys, json, math, os, random, time
from collections import deque
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

# ── 1. Signal generation ──
print("1. Signal Generation Tests")
print("=" * 40)

# Test: Mean Reversion signal logic (extracted from live/node.py)
# Simulate ETH prices with sharp drop
random.seed(42)
eth_prices = deque(maxlen=60)
base = 1800.0
for _ in range(19):
    eth_prices.append(base + random.uniform(-5, 5))
eth_prices.append(base - 20.0)  # sharp -2σ drop

mr_signals = []
w = list(eth_prices)[-20:]
eth_mr = eth_prices[-1]
prior = w[:-1]
sma = sum(prior) / len(prior)
vstd = math.sqrt(sum((p - sma)**2 for p in prior) / len(prior))
dev = (eth_mr - sma) / vstd if vstd > 0 else 0
if dev > 1.0:
    mr_signals.append({"signal": "SELL", "strength": dev})
elif dev < -1.0:
    mr_signals.append({"signal": "BUY", "strength": abs(dev)})

assert len(mr_signals) > 0, f"Mean Reversion should fire on -2σ drop, got 0"
assert mr_signals[0]["signal"] == "BUY", f"Sharp drop below mean should trigger BUY, got {mr_signals[0]}"
print(f"  ✅ Mean Reversion: {mr_signals[0]['signal']} at dev={mr_signals[0]['strength']:.2f}")

# Test: Momentum breakout (Bollinger)
w = [100, 101, 102, 103, 104, 105, 106, 107, 108, 109] + [115, 116, 117, 118, 119, 120, 121, 122, 123, 124]
eth_cur = w[-1]
sma = sum(w) / len(w)
std = math.sqrt(sum((p - sma)**2 for p in w) / len(w))
assert eth_cur > sma + 1.2 * std, f"Expected breakout above 1.2σ band"
print(f"  ✅ Momentum: price {eth_cur} > band {sma + 1.2*std:.1f} — BUY signal")

# Test: Pairs ratio deviation
btc_prices = deque([64000 + i * 100 for i in range(20)], maxlen=60)
eth_prices = deque([1800.0] * 20, maxlen=60)
ratios = [btc_prices[i] / eth_prices[i] for i in range(-20, 0)]
mu = sum(ratios) / len(ratios)
std = math.sqrt(sum((r - mu)**2 for r in ratios) / len(ratios))
cur = btc_prices[-1] / eth_prices[-1]
z = (cur - mu) / std if std > 0 else 0
assert z > 1.2, f"BTC rising vs flat ETH should produce z>1.2, got {z:.2f}"
print(f"  ✅ Pairs Trading: z={z:.2f} — SELL_ETH signal")

# Test: OBI reversal detection
btc_list = list(btc_prices)
ret = (btc_list[-1] - btc_list[-5]) / btc_list[-5]
assert ret > 0.0004, f"5-tick return should be >0.04% on uptrend"
print(f"  ✅ OBI: 5-tick return {ret*100:.2f}% — SELL (overbought)")

# ── 2. Backtest Validation ──
print("\n2. Backtest Validation")
print("=" * 40)

import numpy as np
np.random.seed(7)
n = 500
prices = np.cumsum(np.random.randn(n) * 0.01) + 0.35

equity = 100.0; pos = 0; entry = 0; trades = 0; won = 0
WINDOW = 20
for i in range(WINDOW + 1, n):
    prior = prices[i - WINDOW - 1:i - 1]
    mu = float(np.mean(prior))
    sd = float(np.std(prior, ddof=1))
    z = (prices[i] - mu) / sd if sd > 0 else 0
    if pos == 0:
        if z > 1.5: pos = -1; entry = prices[i]
        elif z < -1.5: pos = 1; entry = prices[i]
    elif pos != 0 and (abs(z) < 0.3):
        pnl = (prices[i] / entry - 1) * pos * equity * 0.01
        equity += pnl; trades += 1
        if pnl > 0: won += 1; pos = 0

pct = (equity / 100.0 - 1) * 100
assert trades > 0, f"Backtest should produce trades on 500-point series"
assert won > 0, f"Should have winning trades, got {won}/{trades}"
print(f"  ✅ SPX MR: ${equity:.2f} ({pct:+.2f}%) | {trades} trades | {won/trades*100:.0f}% win")

# ── 3. Hurst/VPIN ──
print("\n3. Hurst/VPIN Strategy")
print("=" * 40)

from strategies.hurst_vpin import HurstVPINSignal
np.random.seed(1)
n = 2000
trend = np.cumsum(np.random.randn(n) * 50 + 10) + 63000
sides = ['B' if random.random() < 0.65 else 'A' for _ in range(n)]
trade_data = [{"px": float(trend[i]), "sz": 0.01, "side": sides[i]} for i in range(n)]

sg = HurstVPINSignal(notional_threshold=5000.0)
signals = 0
for t in trade_data:
    r = sg.add_trade(t["px"], t["sz"], t["side"])
    if r and r["signal"] != "HOLD":
        signals += 1

assert signals > 0, f"No signals from Hurst/VPIN on trending data"
assert sg.bar_count >= 50, f"Should build 50+ dollar bars, got {sg.bar_count}"
print(f"  ✅ Hurst/VPIN: {signals} signals, {sg.bar_count} dollar bars")

# ── 4. Memory guard ──
print("\n4. Memory Guard")
print("=" * 40)

# Test memory guard independently (don't import server.py — has hardcoded paths)
import gc
import os as _os
MEM_SOFT_LIMIT = 256 * 1024 * 1024
MEM_HARD_LIMIT = 512 * 1024 * 1024

def check_memory():
    try:
        with open("/proc/self/status") as f:
            for line in f:
                if line.startswith("VmRSS:"):
                    rss_kb = int(line.split()[1])
                    rss = rss_kb * 1024
                    if rss > MEM_HARD_LIMIT:
                        _os._exit(1)
                    if rss > MEM_SOFT_LIMIT:
                        gc.collect()
                    return
    except Exception:
        pass

check_memory()  # Should not throw
assert MEM_HARD_LIMIT == 512 * 1024 * 1024
assert MEM_SOFT_LIMIT == 256 * 1024 * 1024
print(f"  ✅ Memory guard: soft={MEM_SOFT_LIMIT//1024//1024}MB hard={MEM_HARD_LIMIT//1024//1024}MB")

# ── 5. Dashboard API (optional) ──
print("\n5. Dashboard API")
print("=" * 40)
try:
    import requests
    r = requests.get("https://ftdt.io/cv/api/backtests/historical", timeout=10)
    assert r.status_code == 200
    data = r.json()
    assert len(data) >= 33, f"Expected 33+ backtests, got {len(data)}"
    spx = [x for x in data if x["strategy"] == "SPX Mean Reversion"]
    assert len(spx) >= 1
    print(f"  ✅ Historical API: {len(data)} backtests ({len(spx)} SPX)")
except Exception as e:
    print(f"  ⚠️  API unreachable: {e}")

# ── 6. Summary ──
print("\n" + "=" * 40)
print("ALL TESTS PASSED ✅")
print("=" * 40)