diff schwab_trader/dashboard_test.py @ 220:eb8b4230fdb9

[schwab-trader] Add guarded trading experiment
author MrJuneJune <me@mrjunejune.com>
date Sun, 02 Aug 2026 08:52:13 -0700
parents
children
line wrap: on
line diff
--- /dev/null	Thu Jan 01 00:00:00 1970 +0000
+++ b/schwab_trader/dashboard_test.py	Sun Aug 02 08:52:13 2026 -0700
@@ -0,0 +1,61 @@
+import json
+import tempfile
+import threading
+import unittest
+import urllib.request
+from http.server import ThreadingHTTPServer
+from pathlib import Path
+
+from schwab_trader.dashboard import DashboardStore, EvidenceInput, score_sentiment
+from schwab_trader.dashboard_server import create_handler
+
+
+class DashboardStoreTest(unittest.TestCase):
+    def test_sentiment_scoring(self):
+        self.assertGreater(score_sentiment("$AAPL bullish strong growth"), 0)
+        self.assertLess(score_sentiment("$AAPL bearish weak sell"), 0)
+
+    def test_evidence_recomputes_signal(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = DashboardStore(Path(temp_dir) / "dashboard.db")
+            store.update_settings({"min_evidence_count": 2, "min_source_count": 2, "min_confidence": 0.3})
+
+            store.add_evidence(EvidenceInput(source="reddit", symbol="AAPL", text="$AAPL bullish strong", engagement=20))
+            result = store.add_evidence(EvidenceInput(source="x", symbol="AAPL", text="$AAPL growth surge", engagement=50))
+
+            self.assertEqual("AAPL", result["symbol"])
+            signals = store.list_signals()
+            self.assertEqual(1, len(signals))
+            self.assertEqual("CONSIDER_BUY", signals[0]["action"])
+
+    def test_paper_trade_rejects_above_max_trade_dollars(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = DashboardStore(Path(temp_dir) / "dashboard.db")
+            store.update_settings({"max_trade_dollars": 100})
+
+            trade = store.add_paper_trade(
+                {"symbol": "AAPL", "action": "BUY", "quantity": 2, "price": 75, "reason": "test"}
+            )
+
+            self.assertEqual("rejected_max_trade_dollars", trade["status"])
+
+    def test_dashboard_http_status(self):
+        with tempfile.TemporaryDirectory() as temp_dir:
+            store = DashboardStore(Path(temp_dir) / "dashboard.db")
+            server = ThreadingHTTPServer(("127.0.0.1", 0), create_handler(store))
+            thread = threading.Thread(target=server.serve_forever, daemon=True)
+            thread.start()
+            try:
+                url = f"http://127.0.0.1:{server.server_port}/api/status"
+                with urllib.request.urlopen(url, timeout=5) as response:
+                    body = json.loads(response.read().decode("utf-8"))
+                self.assertEqual("schwab-dashboard", body["service"])
+                self.assertFalse(body["live_trading_enabled"])
+            finally:
+                server.shutdown()
+                thread.join(timeout=5)
+
+
+if __name__ == "__main__":
+    unittest.main()
+