#!/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)