import json import os import sqlite3 import subprocess import sys import tempfile import unittest from datetime import datetime,timedelta,timezone from decimal import Decimal from pathlib import Path from unittest.mock import patch ROOT=Path(__file__).resolve().parents[1] sys.path.insert(0,str(ROOT/'runtime')) from engine import Engine,Rejected,holding_class,number,timestamp,volume_breaker,CHANNELS from server import Server from alpaca_reader import paper_account_check def reasoning(): now=timestamp() return {'decision':'RESEARCH_PASSED','analyzed_at':now,**{ k:{'status':'PASS','summary':'Synthetic test evidence only', 'sources':[{'url':'https://example.org/test-fixture','observed_at':now}]} for k in CHANNELS}} def proposal(symbol='BTC/USD',entry='100',stop='95'): return dict(symbol=symbol,entry=entry,stop=stop,quote_at=timestamp(),reasoning=reasoning()) class EngineTests(unittest.TestCase): def setUp(self): self.engine=Engine(':memory:') self.engine.initialize('25000') def tearDown(self): self.engine.db.close() def test_dynamic_risk_and_cash(self): e=self.engine.evaluate(**proposal()) self.assertLessEqual(Decimal(e['planned_risk']),Decimal('250')) self.assertLessEqual(Decimal(e['notional']),Decimal('6250')) self.engine.db.execute("UPDATE accounts SET cash='1000',initial_equity='1000',session_equity='1000',peak_equity='1000'") e=self.engine.evaluate(**proposal()) self.assertEqual(e['risk_cap'],'10.00') def test_cash_downsizes(self): self.engine.db.execute("UPDATE accounts SET cash='100',initial_equity='100',session_equity='100',peak_equity='100'") e=self.engine.evaluate(**proposal(stop='1')) self.assertLessEqual(Decimal(e['notional'])+Decimal(e['entry_fee']),Decimal('100')) def test_sector_combined(self): e=self.engine.simulate_entry('a',**proposal(stop='99')) self.assertLessEqual(Decimal(e['notional']),self.engine.equity()*Decimal('.25')) with self.assertRaises(Rejected): self.engine.simulate_entry('b',**proposal('ETH/USD',stop='99')) def test_duplicate_atomic(self): p=proposal() a=self.engine.simulate_entry('same',**p) b=self.engine.simulate_entry('same',**p) self.assertEqual(a['order_id'],b['order_id']) self.assertEqual(self.engine.db.execute('SELECT COUNT(*) FROM reasons').fetchone()[0],1) with self.assertRaises(Rejected): self.engine.simulate_entry('same',**proposal(entry='101')) def test_rollback_reason_failure(self): self.engine.db.execute("CREATE TRIGGER reject_reason BEFORE INSERT ON reasons BEGIN SELECT RAISE(ABORT,'fixture'); END;") with self.assertRaises(sqlite3.IntegrityError): self.engine.simulate_entry('x',**proposal()) self.assertEqual(self.engine.db.execute('SELECT COUNT(*) FROM orders').fetchone()[0],0) self.assertEqual(self.engine.account()['cash'],'25000') def test_no_fabricated_research(self): p=proposal();p['reasoning'].pop('tax') with self.assertRaises(Rejected): self.engine.evaluate(**p) def test_source_required(self): p=proposal();p['reasoning']['tax']['sources']=[] with self.assertRaises(Rejected): self.engine.evaluate(**p) def test_expired_research(self): p=proposal();p['reasoning']['analyzed_at']=(datetime.now(timezone.utc)-timedelta(hours=2)).isoformat() with self.assertRaises(Rejected): self.engine.evaluate(**p) def test_invalid_numbers(self): for x in ('NaN','Infinity','-1','invalid'): with self.subTest(x=x),self.assertRaises(Rejected): number(x) def test_price_and_stop(self): for e,s in [('0','1'),('100','100'),('100','101')]: with self.subTest(e=e,s=s),self.assertRaises(Rejected): self.engine.evaluate(**proposal(entry=e,stop=s)) def test_universe(self): with self.assertRaises(Rejected): self.engine.evaluate(**proposal('SOL/USD')) with self.assertRaises(Rejected): self.engine.evaluate(**proposal('/MES')) def test_future_and_stale(self): for delta in (-120,60): p=proposal();p['quote_at']=(datetime.now(timezone.utc)+timedelta(seconds=delta)).isoformat() with self.assertRaises(Rejected): self.engine.evaluate(**p) def test_gap_exceeds_planned_stop_risk(self): e=self.engine.simulate_entry('gap',**proposal()) r=self.engine.mark({'BTC/USD':{'price':'50','at':timestamp()}}) self.assertEqual(len(r['exits']),1) loss=-Decimal(r['exits'][0]['economic_pnl']) self.assertGreater(loss,Decimal(e['risk_cap'])) self.assertEqual(r['status'],'HALTED') self.assertTrue(self.engine.account()['halted']) def test_complete_mark_atomic(self): self.engine.simulate_entry('a',**proposal()) before=self.engine.positions()[0]['mark'] with self.assertRaises(Rejected): self.engine.mark({}) self.assertEqual(self.engine.positions()[0]['mark'],before) def test_stop_disposition_and_year(self): self.engine.simulate_entry('stop',**proposal()) self.engine.mark({'BTC/USD':{'price':'94','at':timestamp()}}) self.assertEqual(self.engine.report()['closed_lots'],1) self.assertEqual(self.engine.tax_export(datetime.now(timezone.utc).year-1)['records'],[]) out=self.engine.tax_export(datetime.now(timezone.utc).year) self.assertEqual(len(out['records']),1) self.assertFalse(out['filed']) def test_no_short(self): with self.assertRaises(Rejected): self.engine.close_position('BTC/USD','100',timestamp()) def test_no_loss_pf_undefined(self): r=self.engine.report() self.assertIsNone(r['profit_factor']) self.assertEqual(r['closed_lots'],0) def test_no_automatic_live_clearance(self): self.assertEqual(self.engine.audit()['status'],'REVIEW_REQUIRED_LIVE_DISABLED') self.assertEqual(self.engine.audit()['trial_days'],0) def test_live_config_rejected(self): for env in ({'TRADING_MODE':'LIVE'},{'ALPACA_BASE_URL':'https://api.alpaca.markets'}): with patch.dict(os.environ,env),self.assertRaises(Rejected): Engine(':memory:') def test_no_credential_fallback(self): with patch.dict(os.environ,{'ALPACA_PAPER_KEY':'AK_MOCK','ALPACA_PAPER_SECRET':'mock'}),self.assertRaises(Rejected): paper_account_check() def test_no_reset(self): with self.assertRaises(Rejected): self.engine.initialize('100') def test_heartbeat_idle_not_trade(self): self.engine.heartbeat() self.assertEqual(self.engine.status()['workers'][0]['status'],'IDLE') self.assertTrue(self.engine.status()['workers'][0]['fresh']) self.assertEqual(self.engine.db.execute('SELECT COUNT(*) FROM orders').fetchone()[0],0) def test_calendar_year_leap(self): self.assertEqual(holding_class('2023-03-01','2024-03-01'),'SHORT_TERM') self.assertEqual(holding_class('2023-03-01','2024-03-02'),'LONG_TERM') with self.assertRaises(Rejected): holding_class('2024-01-02','2024-01-01') def test_volume_constant_halts(self): self.assertEqual(volume_breaker(['100']*20,'1000')['action'],'HALT_UNCALIBRATED_BASELINE') def test_volume_threshold(self): h=[str(100+x) for x in range(20)] self.assertEqual(volume_breaker(h,'1000')['action'],'HALT_REVIEW') with self.assertRaises(Rejected): volume_breaker(h[:19],'1000') def test_recovery_same_session_rejected(self): with self.assertRaises(Rejected): self.engine.recover() def test_halted_next_day_can_close(self): self.engine.simulate_entry('a',**proposal()) yesterday=(datetime.now(timezone.utc)-timedelta(days=1)).date().isoformat() self.engine.db.execute('UPDATE accounts SET halted=1,session=?',(yesterday,)) result=self.engine.mark({'BTC/USD':{'price':'100','at':timestamp()}}) self.assertEqual(result['status'],'HALTED') self.assertEqual(len(self.engine.positions()),0) self.engine.recover() self.assertFalse(self.engine.account()['halted']) def test_latched_halt_rejects_entries(self): self.engine.db.execute('UPDATE accounts SET halted=1') with self.assertRaises(Rejected): self.engine.simulate_entry('halt',**proposal()) def test_utc_required(self): p=proposal();p['quote_at']='2026-10-04T08:00:00' with self.assertRaises(Rejected): self.engine.evaluate(**p) def test_backup_restore(self): self.engine.simulate_entry('backup',**proposal()) with tempfile.TemporaryDirectory() as d: path=str(Path(d)/'backup.sqlite3') destination=sqlite3.connect(path) self.engine.db.backup(destination) destination.close() restored=Engine(path) self.assertEqual(restored.db.execute('PRAGMA integrity_check').fetchone()[0],'ok') self.assertEqual(restored.account()['cash'],self.engine.account()['cash']) self.assertEqual(restored.positions()[0]['symbol'],'BTC/USD') self.assertEqual(restored.db.execute('SELECT COUNT(*) FROM reasons').fetchone()[0],1) restored.db.close() class ProtocolTests(unittest.TestCase): def test_real_stdio_startup_discovery_and_call(self): requests=[{'jsonrpc':'2.0','id':1,'method':'initialize','params':{'protocolVersion':'2025-03-26','capabilities':{},'clientInfo':{'name':'test','version':'1'}}}, {'jsonrpc':'2.0','method':'notifications/initialized'}, {'jsonrpc':'2.0','id':2,'method':'tools/list'}, {'jsonrpc':'2.0','id':3,'method':'tools/call','params':{'name':'quant_status','arguments':{}}}, {'jsonrpc':'2.0','id':4,'method':'tools/call','params':{'name':'initialize_simulation','arguments':{'virtual_cash':'25000'}}}, {'jsonrpc':'2.0','id':5,'method':'tools/call','params':{'name':'performance_report','arguments':{'days':True}}}] with tempfile.TemporaryDirectory() as d: env={**os.environ,'QUANT_DB_PATH':str(Path(d)/'db.sqlite3'),'TRADING_MODE':'LOCAL_SIMULATION','ALPACA_BASE_URL':'https://paper-api.alpaca.markets'} result=subprocess.run([sys.executable,str(ROOT/'runtime/server.py')],input='\n'.join(json.dumps(x) for x in requests)+'\n',text=True,capture_output=True,env=env,check=True) replies=[json.loads(x) for x in result.stdout.splitlines()] self.assertEqual(len(replies),5) self.assertEqual(len(replies[1]['result']['tools']),12) status=json.loads(replies[2]['result']['content'][0]['text']) self.assertFalse(status['live_execution_enabled']) self.assertFalse(status['account_initialized']) self.assertFalse(replies[3]['result']['isError']) self.assertTrue(replies[4]['result']['isError']) self.assertEqual(result.stderr,'') if __name__=='__main__': unittest.main()