303 lines
13 KiB
Python
303 lines
13 KiB
Python
#!/usr/bin/env python
|
|
"""Offline tests — no IB connection, no orders, safe to run against the live repo.
|
|
|
|
.venv/bin/python -m unittest test_offline -v
|
|
|
|
Covers the pure layers: ledger writing (PositionTracker), round-trip
|
|
reconstruction and statistics (analytics), and the indicator maths. Indicator
|
|
tests are skipped when pandas is unavailable.
|
|
"""
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import analytics
|
|
from state import PositionTracker
|
|
|
|
try:
|
|
import pandas as pd
|
|
HAVE_PANDAS = True
|
|
except ImportError:
|
|
HAVE_PANDAS = False
|
|
|
|
|
|
class LedgerTest(unittest.TestCase):
|
|
"""PositionTracker must record the attribution fields the dashboard needs."""
|
|
|
|
def setUp(self):
|
|
self.dir = tempfile.TemporaryDirectory()
|
|
self.tracker = PositionTracker(str(Path(self.dir.name) / "bot_state.json"))
|
|
|
|
def tearDown(self):
|
|
self.dir.cleanup()
|
|
|
|
def _ledger(self):
|
|
return [json.loads(l) for l in
|
|
self.tracker.ledger_file.read_text().splitlines() if l.strip()]
|
|
|
|
def test_buy_records_reason_and_commission(self):
|
|
self.tracker.record_buy("S", "AAPL", 10, 100.0, reason="RSI", commission=0.35)
|
|
rec = self._ledger()[-1]
|
|
self.assertEqual(rec["type"], "buy")
|
|
self.assertEqual(rec["reason"], "RSI")
|
|
self.assertEqual(rec["commission"], 0.35)
|
|
|
|
def test_sell_records_reason_and_commission(self):
|
|
self.tracker.record_buy("S", "AAPL", 10, 100.0, reason="RSI")
|
|
self.tracker.record_sell("S", "AAPL", 10, 103.0,
|
|
reason="SignalExit", commission=0.4)
|
|
rec = self._ledger()[-1]
|
|
self.assertEqual(rec["reason"], "SignalExit")
|
|
self.assertEqual(rec["commission"], 0.4)
|
|
|
|
def test_absent_fields_are_omitted(self):
|
|
"""A record with no reason/commission stays byte-compatible with the old
|
|
schema, so historical ledgers keep parsing."""
|
|
self.tracker.record_buy("S", "AAPL", 10, 100.0)
|
|
rec = self._ledger()[-1]
|
|
self.assertNotIn("reason", rec)
|
|
self.assertNotIn("commission", rec)
|
|
|
|
def test_day_pnl_is_net_of_commission(self):
|
|
self.tracker.record_buy("S", "AAPL", 10, 100.0)
|
|
self.tracker.record_sell("S", "AAPL", 10, 101.0, commission=0.5)
|
|
self.assertAlmostEqual(self.tracker.day_realized_pnl(), 10.0 - 0.5, places=6)
|
|
|
|
def test_external_sell_is_tagged(self):
|
|
self.tracker.record_buy("S", "AAPL", 10, 100.0)
|
|
self.tracker.reconcile({}) # account holds none -> backfill an external sell
|
|
rec = self._ledger()[-1]
|
|
self.assertEqual(rec["type"], "sell_external")
|
|
self.assertEqual(rec["reason"], "HardStopOffline")
|
|
self.assertTrue(rec["est"])
|
|
|
|
|
|
class RoundTripTest(unittest.TestCase):
|
|
"""FIFO reconstruction has to survive partial fills and pre-ledger lots."""
|
|
|
|
@staticmethod
|
|
def _trips(records):
|
|
return analytics.build_round_trips(records)
|
|
|
|
def test_simple_round_trip(self):
|
|
trips, open_lots, unmatched = self._trips([
|
|
{"ts": "2026-08-01T10:00:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 10, "price": 100.0, "reason": "Entry", "commission": 0.35},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 10, "price": 102.0, "reason": "SignalExit", "commission": 0.35},
|
|
])
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(open_lots, [])
|
|
self.assertEqual(unmatched, 0)
|
|
t = trips[0]
|
|
self.assertAlmostEqual(t.gross_pnl, 20.0)
|
|
self.assertAlmostEqual(t.commission, 0.70)
|
|
self.assertAlmostEqual(t.pnl, 19.30)
|
|
self.assertAlmostEqual(t.pnl_pct, 19.30 / 1000 * 100)
|
|
self.assertEqual(t.hold_minutes, 60)
|
|
self.assertEqual(t.entry_reason, "Entry")
|
|
self.assertEqual(t.exit_reason, "SignalExit")
|
|
self.assertTrue(t.won)
|
|
|
|
def test_partial_sell_splits_the_lot(self):
|
|
trips, open_lots, _ = self._trips([
|
|
{"ts": "2026-08-01T10:00:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 10, "price": 100.0, "commission": 1.0},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 4, "price": 105.0},
|
|
])
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0].qty, 4)
|
|
self.assertAlmostEqual(trips[0].gross_pnl, 20.0)
|
|
# buy commission is allocated pro rata: 4/10 of $1.00
|
|
self.assertAlmostEqual(trips[0].commission, 0.4)
|
|
self.assertEqual(len(open_lots), 1)
|
|
self.assertEqual(open_lots[0].qty, 6)
|
|
|
|
def test_sell_spanning_two_lots_is_fifo(self):
|
|
trips, _, _ = self._trips([
|
|
{"ts": "2026-08-01T10:00:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 5, "price": 100.0},
|
|
{"ts": "2026-08-01T10:30:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 5, "price": 110.0},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 8, "price": 120.0},
|
|
])
|
|
self.assertEqual(len(trips), 2)
|
|
# oldest lot consumed first, in full
|
|
self.assertEqual((trips[0].qty, trips[0].entry_price), (5, 100.0))
|
|
self.assertEqual((trips[1].qty, trips[1].entry_price), (3, 110.0))
|
|
self.assertAlmostEqual(sum(t.gross_pnl for t in trips), 5 * 20 + 3 * 10)
|
|
|
|
def test_sell_without_basis_is_reported_not_guessed(self):
|
|
trips, _, unmatched = self._trips([
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 7, "price": 120.0},
|
|
])
|
|
self.assertEqual(trips, [])
|
|
self.assertEqual(unmatched, 7)
|
|
|
|
def test_positions_are_owned_per_strategy(self):
|
|
"""Two strategies holding the same symbol must not cross-match."""
|
|
trips, open_lots, unmatched = self._trips([
|
|
{"ts": "2026-08-01T10:00:00", "type": "buy", "strategy": "A", "symbol": "X",
|
|
"qty": 5, "price": 100.0},
|
|
{"ts": "2026-08-01T10:01:00", "type": "buy", "strategy": "B", "symbol": "X",
|
|
"qty": 5, "price": 200.0},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "B", "symbol": "X",
|
|
"qty": 5, "price": 210.0},
|
|
])
|
|
self.assertEqual(len(trips), 1)
|
|
self.assertEqual(trips[0].strategy, "B")
|
|
self.assertAlmostEqual(trips[0].gross_pnl, 50.0)
|
|
self.assertEqual(unmatched, 0)
|
|
self.assertEqual([(l.strategy, l.qty) for l in open_lots], [("A", 5)])
|
|
|
|
def test_seed_lot_is_flagged(self):
|
|
trips, _, _ = self._trips([
|
|
{"ts": "2026-08-01T10:00:00", "type": "seed", "strategy": "S", "symbol": "X",
|
|
"qty": 5, "price": 100.0},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 5, "price": 101.0},
|
|
])
|
|
self.assertTrue(trips[0].seeded_entry)
|
|
|
|
|
|
class StatsTest(unittest.TestCase):
|
|
"""The statistics that parameter decisions get made on."""
|
|
|
|
@staticmethod
|
|
def _trip(pnl, entry=100.0, qty=1.0, ts="2026-08-01T10:00:00"):
|
|
return analytics.RoundTrip(
|
|
strategy="S", symbol="X", qty=qty, entry_ts=ts, exit_ts=ts,
|
|
entry_price=entry, exit_price=entry + pnl, entry_reason="", exit_reason="",
|
|
gross_pnl=pnl, commission=0.0, estimated=False, seeded_entry=False)
|
|
|
|
def test_summary_of_empty_set(self):
|
|
s = analytics.summarize([])
|
|
self.assertEqual(s["n"], 0)
|
|
self.assertEqual(s["pnl"], 0.0)
|
|
self.assertIsNone(s["profit_factor"])
|
|
|
|
def test_core_statistics(self):
|
|
# 3 wins of +10, 2 losses of -20
|
|
trips = [self._trip(10)] * 3 + [self._trip(-20)] * 2
|
|
s = analytics.summarize(trips)
|
|
self.assertEqual(s["n"], 5)
|
|
self.assertAlmostEqual(s["pnl"], 30 - 40)
|
|
self.assertAlmostEqual(s["win_rate"], 60.0)
|
|
self.assertAlmostEqual(s["expectancy"], -2.0)
|
|
self.assertAlmostEqual(s["profit_factor"], 30 / 40)
|
|
self.assertAlmostEqual(s["avg_win"], 10.0)
|
|
self.assertAlmostEqual(s["avg_loss"], 20.0)
|
|
self.assertAlmostEqual(s["payoff_ratio"], 0.5)
|
|
# payoff 0.5 => must win 2 of every 3 just to break even
|
|
self.assertAlmostEqual(s["breakeven_win_rate"], 100 / 1.5)
|
|
|
|
def test_breakeven_win_rate_matches_actual_at_zero_expectancy(self):
|
|
"""Sanity check on the headline diagnostic: when a set nets exactly zero,
|
|
its win rate must equal its own breakeven win rate."""
|
|
trips = [self._trip(10)] * 2 + [self._trip(-20)]
|
|
s = analytics.summarize(trips)
|
|
self.assertAlmostEqual(s["pnl"], 0.0)
|
|
self.assertAlmostEqual(s["win_rate"], s["breakeven_win_rate"], places=6)
|
|
|
|
def test_profit_factor_is_none_without_losses(self):
|
|
self.assertIsNone(analytics.summarize([self._trip(5)])["profit_factor"])
|
|
|
|
def test_max_drawdown(self):
|
|
trips = [
|
|
self._trip(100, ts="2026-08-01T10:00:00"),
|
|
self._trip(-40, ts="2026-08-01T11:00:00"),
|
|
self._trip(-30, ts="2026-08-01T12:00:00"),
|
|
self._trip(50, ts="2026-08-01T13:00:00"),
|
|
]
|
|
curve = analytics.equity_curve(trips)
|
|
self.assertEqual([p["cum_pnl"] for p in curve], [100, 60, 30, 80])
|
|
self.assertAlmostEqual(analytics.max_drawdown(curve), -70.0)
|
|
|
|
def test_grouping(self):
|
|
a = self._trip(10); a.symbol = "AAA"
|
|
b = self._trip(-5); b.symbol = "BBB"
|
|
g = analytics.group_stats([a, b], lambda t: t.symbol)
|
|
self.assertAlmostEqual(g["AAA"]["pnl"], 10)
|
|
self.assertAlmostEqual(g["BBB"]["pnl"], -5)
|
|
|
|
def test_analyze_end_to_end_with_date_filter(self):
|
|
recs = [
|
|
{"ts": "2026-08-01T10:00:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 1, "price": 100.0},
|
|
{"ts": "2026-08-01T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 1, "price": 110.0},
|
|
{"ts": "2026-08-05T10:00:00", "type": "buy", "strategy": "S", "symbol": "X",
|
|
"qty": 1, "price": 100.0},
|
|
{"ts": "2026-08-05T11:00:00", "type": "sell", "strategy": "S", "symbol": "X",
|
|
"qty": 1, "price": 90.0},
|
|
]
|
|
with tempfile.TemporaryDirectory() as d:
|
|
p = Path(d) / "trades.jsonl"
|
|
p.write_text("\n".join(json.dumps(r) for r in recs) + "\n")
|
|
allt = analytics.analyze(p)
|
|
self.assertEqual(allt["overall"]["n"], 2)
|
|
self.assertAlmostEqual(allt["overall"]["pnl"], 0.0)
|
|
# filtering the window must not corrupt the cost basis of what remains
|
|
late = analytics.analyze(p, since="2026-08-05")
|
|
self.assertEqual(late["overall"]["n"], 1)
|
|
self.assertAlmostEqual(late["overall"]["pnl"], -10.0)
|
|
|
|
|
|
@unittest.skipUnless(HAVE_PANDAS, "pandas not installed")
|
|
class IndicatorTest(unittest.TestCase):
|
|
"""Guards on the two indicator bugs AGENTS.md warns against reintroducing."""
|
|
|
|
def test_rsi_is_100_on_a_pure_uptrend(self):
|
|
from strategies.mean_reversion import MeanReversionStrategy
|
|
rsi = MeanReversionStrategy._calc_rsi(pd.Series(range(1, 40), dtype=float), 14)
|
|
# no losses -> rs is inf -> RSI must saturate at 100, never NaN
|
|
self.assertFalse(pd.isna(rsi.iloc[-1]))
|
|
self.assertAlmostEqual(rsi.iloc[-1], 100.0, places=6)
|
|
|
|
def test_adx_direction_on_a_clean_uptrend(self):
|
|
from strategies.ma_cross import MAStockStrategy
|
|
n = 60
|
|
close = pd.Series([100 + i for i in range(n)], dtype=float)
|
|
df = pd.DataFrame({"high": close + 0.5, "low": close - 0.5, "close": close})
|
|
adx = MAStockStrategy._calc_adx(df, 14)
|
|
# a monotonic uptrend must read as strongly trending
|
|
self.assertGreater(adx.iloc[-1], 50)
|
|
|
|
def test_vwap_resets_each_day(self):
|
|
from strategies.short_term import ShortTermMAVWAPStrategy
|
|
ts = (list(pd.date_range("2026-08-03 13:30", periods=3, freq="1min", tz="UTC"))
|
|
+ list(pd.date_range("2026-08-04 13:30", periods=3, freq="1min", tz="UTC")))
|
|
df = pd.DataFrame({
|
|
"date": ts,
|
|
"high": [10, 10, 10, 20, 20, 20], "low": [10, 10, 10, 20, 20, 20],
|
|
"close": [10, 10, 10, 20, 20, 20], "volume": [1, 1, 1, 1, 1, 1],
|
|
})
|
|
vwap = ShortTermMAVWAPStrategy._calc_vwap(df)
|
|
# day two must not be dragged toward day one's prices
|
|
self.assertAlmostEqual(vwap.iloc[2], 10.0)
|
|
self.assertAlmostEqual(vwap.iloc[3], 20.0)
|
|
|
|
def test_forming_bar_is_dropped(self):
|
|
from datetime import datetime, timedelta, timezone
|
|
from ib_insync import BarData
|
|
from bars import to_completed_df
|
|
|
|
def bar(date, close):
|
|
return BarData(date=date, open=close, high=close, low=close,
|
|
close=close, volume=1)
|
|
|
|
now = datetime.now(timezone.utc)
|
|
bars = [bar(now - timedelta(seconds=120), 1.0), bar(now - timedelta(seconds=60), 2.0),
|
|
bar(now, 3.0)] # the last bar is still forming
|
|
df = to_completed_df(bars, 60)
|
|
self.assertEqual(len(df), 2)
|
|
self.assertEqual(df["close"].iloc[-1], 2.0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|