Mercurial
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() +