Mercurial

comparison schwab_trader/dashboard_test.py @ 220:eb8b4230fdb9

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[schwab-trader] Add guarded trading experiment
author MrJuneJune <me@mrjunejune.com>
date Sun, 02 Aug 2026 08:52:13 -0700
parents
children
comparison
equal deleted inserted replaced
214:4c725fde6999 220:eb8b4230fdb9
1 import json
2 import tempfile
3 import threading
4 import unittest
5 import urllib.request
6 from http.server import ThreadingHTTPServer
7 from pathlib import Path
8
9 from schwab_trader.dashboard import DashboardStore, EvidenceInput, score_sentiment
10 from schwab_trader.dashboard_server import create_handler
11
12
13 class DashboardStoreTest(unittest.TestCase):
14 def test_sentiment_scoring(self):
15 self.assertGreater(score_sentiment("$AAPL bullish strong growth"), 0)
16 self.assertLess(score_sentiment("$AAPL bearish weak sell"), 0)
17
18 def test_evidence_recomputes_signal(self):
19 with tempfile.TemporaryDirectory() as temp_dir:
20 store = DashboardStore(Path(temp_dir) / "dashboard.db")
21 store.update_settings({"min_evidence_count": 2, "min_source_count": 2, "min_confidence": 0.3})
22
23 store.add_evidence(EvidenceInput(source="reddit", symbol="AAPL", text="$AAPL bullish strong", engagement=20))
24 result = store.add_evidence(EvidenceInput(source="x", symbol="AAPL", text="$AAPL growth surge", engagement=50))
25
26 self.assertEqual("AAPL", result["symbol"])
27 signals = store.list_signals()
28 self.assertEqual(1, len(signals))
29 self.assertEqual("CONSIDER_BUY", signals[0]["action"])
30
31 def test_paper_trade_rejects_above_max_trade_dollars(self):
32 with tempfile.TemporaryDirectory() as temp_dir:
33 store = DashboardStore(Path(temp_dir) / "dashboard.db")
34 store.update_settings({"max_trade_dollars": 100})
35
36 trade = store.add_paper_trade(
37 {"symbol": "AAPL", "action": "BUY", "quantity": 2, "price": 75, "reason": "test"}
38 )
39
40 self.assertEqual("rejected_max_trade_dollars", trade["status"])
41
42 def test_dashboard_http_status(self):
43 with tempfile.TemporaryDirectory() as temp_dir:
44 store = DashboardStore(Path(temp_dir) / "dashboard.db")
45 server = ThreadingHTTPServer(("127.0.0.1", 0), create_handler(store))
46 thread = threading.Thread(target=server.serve_forever, daemon=True)
47 thread.start()
48 try:
49 url = f"http://127.0.0.1:{server.server_port}/api/status"
50 with urllib.request.urlopen(url, timeout=5) as response:
51 body = json.loads(response.read().decode("utf-8"))
52 self.assertEqual("schwab-dashboard", body["service"])
53 self.assertFalse(body["live_trading_enabled"])
54 finally:
55 server.shutdown()
56 thread.join(timeout=5)
57
58
59 if __name__ == "__main__":
60 unittest.main()
61