#!/usr/bin/env python3 # SPDX-License-Identifier: MIT """AlphaBot V2: stdlib CLI, streamed prices, SQLite macro/events and durable risk state. Live mode speaks the documented AlphaBot gateway contract, never a broker directly. """ import argparse import asyncio import base64 from concurrent.futures import ThreadPoolExecutor from datetime import datetime, timezone from decimal import Decimal, InvalidOperation, getcontext import fcntl import getpass import hashlib import http.cookiejar import json import os from pathlib import Path import platform import resource import re import signal import sqlite3 import ssl import sys import time import threading import urllib.error import urllib.parse import urllib.request import uuid from host_strategy import capital_floor, decide from audit_ledger import Ledger, fixed, dec D = Decimal STOP = False class Fault(Exception): pass def number(value): try: result = D(str(value)) if not result.is_finite(): raise ValueError return result except (InvalidOperation, ValueError): raise Fault('invalid finite numeric value') from None def stamp(value): text = str(value) if len(text) == 10: text += 'T00:00:00+00:00' try: result = datetime.fromisoformat(text.replace('Z', '+00:00')) if result.tzinfo is None: raise ValueError return result.astimezone(timezone.utc) except ValueError: raise Fault('timestamp requires ISO date or timezone-aware ISO timestamp') from None def emit(event, **values): print(json.dumps({'event': event, **values}, default=str, allow_nan=False), flush=True) def records(path): with path.open(encoding='utf-8') as stream: line_no = 0 while True: line = stream.readline(4 * 1024 * 1024 + 1) if not line: break line_no += 1 if len(line) > 4 * 1024 * 1024: raise Fault('JSONL record exceeds 4 MiB') if not line.strip(): continue try: value = json.loads(line) if not isinstance(value, dict): raise ValueError except (ValueError, TypeError): raise Fault(f'malformed JSONL at line {line_no}') from None yield value class Store: def __init__(self, root, mode): self.lock = (root / f'{mode}.lock').open('a') try: fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: self.lock.close() raise Fault('another runner owns this project/mode') from None self.db = sqlite3.connect(root / f'{mode}.sqlite3') self.db.executescript('''PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL; CREATE TABLE IF NOT EXISTS state(k TEXT PRIMARY KEY,v TEXT NOT NULL); CREATE TABLE IF NOT EXISTS macro(obs TEXT, available TEXT, value TEXT, PRIMARY KEY(obs,available)); CREATE INDEX IF NOT EXISTS macro_available ON macro(available); CREATE TABLE IF NOT EXISTS geo(at TEXT PRIMARY KEY,tone TEXT); CREATE TABLE IF NOT EXISTS audit(id INTEGER PRIMARY KEY,body TEXT NOT NULL);''') def get(self, key, default=None): row = self.db.execute('SELECT v FROM state WHERE k=?', (key,)).fetchone() return json.loads(row[0]) if row else default def put(self, **values): own = not self.db.in_transaction if own: self.db.execute('BEGIN IMMEDIATE') try: for key, value in values.items(): self.db.execute('INSERT OR REPLACE INTO state VALUES(?,?)', (key, json.dumps(value, default=str))) if own: self.db.commit() except BaseException: if own: self.db.rollback() raise def audit(self, event, **values): body = {'event': event, **values} own = not self.db.in_transaction if own: self.db.execute('BEGIN IMMEDIATE') try: self.db.execute('INSERT INTO audit(body) VALUES(?)', (json.dumps(fixed(body)),)) if own: self.db.commit() except BaseException: if own: self.db.rollback() raise emit(event, **fixed(values)) def ingest(self, root): sources = [root / 'macro.jsonl', root / 'geopolitical.jsonl'] if any(not p.is_file() for p in sources): raise Fault('macro.jsonl and geopolitical.jsonl are required') signature = tuple((p.stat().st_mtime_ns, p.stat().st_size) for p in sources) if getattr(self, '_ingest_signature', None) == signature: return # Rebuild indexes transactionally; bounded memory regardless of file size. with self.db: self.db.execute('DELETE FROM macro') self.db.execute('DELETE FROM geo') for row in records(root / 'macro.jsonl'): if row['series'] != 'INDPRO': continue obs = stamp(row['observation_date']) supplied = row.get('available_at', row.get('publication_date')) available = obs if obs.year < 2000 or not supplied else stamp(supplied) if available < obs: raise Fault('macro availability precedes observation') self.db.execute('INSERT OR REPLACE INTO macro VALUES(?,?,?)', (obs.isoformat(), available.isoformat(), str(number(row['value'])))) path = root / 'geopolitical.jsonl' if not path.is_file(): raise Fault('geopolitical.jsonl required; empty file explicitly means no events') for row in records(path): at = stamp(row['at']).isoformat() # Preserve simultaneous events by conservatively retaining worst tone. old = self.db.execute('SELECT tone FROM geo WHERE at=?', (at,)).fetchone() tone = number(row['tone']) if old: tone = min(tone, number(old[0])) self.db.execute('INSERT OR REPLACE INTO geo VALUES(?,?)', (at, str(tone))) self._ingest_signature = signature def indicators(self, at): # Select latest visible vintage per month, then demand four consecutive months. months = {} for obs, available, value in self.db.execute( 'SELECT obs,available,value FROM macro WHERE obs<=? AND available<=? ' 'ORDER BY obs DESC,available DESC', (at.isoformat(), at.isoformat())): month = obs[:7] if month not in months: months[month] = number(value) if len(months) == 4: break keys = sorted(months) indices = [int(k[:4]) * 12 + int(k[5:7]) for k in keys] values = [months[k] for k in keys] if len(indices) == 4 and indices[-1] - indices[0] == 3 else [] # Never expose future events merely because their calendar month has begun. tones = (number(r[0]) for r in self.db.execute( 'SELECT tone FROM geo WHERE at>=? AND at<=?', (at.replace(day=1, hour=0, minute=0, second=0, microsecond=0).isoformat(), at.isoformat()))) count = self.db.execute('SELECT count(*) FROM geo WHERE substr(at,1,7)=?', (at.strftime('%Y-%m'),)).fetchone()[0] return values, min(tones, default=D('0') if count == 0 else None), bool(count == 0), keys[-1] if keys else '' class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None class AuthenticationFailed(Fault): def __init__(self): super().__init__('AUTHENTICATION_FAILED') class Gateway: def __init__(self, cfg): self.broker = cfg['broker'] self.account_id = cfg['account_id'] self.deadline = 0.0 self.auth_lock = threading.Lock() self.revoked = threading.Event() self.closed = threading.Event() self.revoked.set() self.cookies = http.cookiejar.CookieJar() if self.broker not in ('alpaca', 'kraken', 'bitstamp'): raise Fault('unsupported broker') self.base = os.environ.get('LIVE_AUTH_GATEWAY_URL', '') if self.base and '://' not in self.base: self.base = 'https://' + self.base parsed = urllib.parse.urlsplit(self.base) if (parsed.scheme != 'https' or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment or parsed.path not in ('', '/')): raise AuthenticationFailed() if parsed.hostname != cfg.get('gateway_hostname'): raise AuthenticationFailed() self.hostname = parsed.hostname self.base = self.base.rstrip('/') self.http = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect(), urllib.request.HTTPCookieProcessor(self.cookies), urllib.request.HTTPSHandler(context=ssl.create_default_context())) if platform.system() != 'Darwin' or not sys.stdin.isatty(): raise AuthenticationFailed() resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) key = os.environ.get('LIVE_BROKER_API_KEY') secret = os.environ.get('LIVE_BROKER_SECRET_KEY') try: if not key or not secret: raise AuthenticationFailed() self.authenticate() # Separate authenticated bootstrap, never extra fields in the gate payload. result = self.call('/v1/broker-session', {'broker': self.broker, 'account_id': self.account_id, 'api_key': key, 'secret_key': secret}) if result.get('bound') is not True or result.get('account_id') != self.account_id: raise AuthenticationFailed() except BaseException: self.cookies.clear() self.revoked.set() raise finally: key = secret = None os.environ.pop('LIVE_BROKER_API_KEY', None) os.environ.pop('LIVE_BROKER_SECRET_KEY', None) self.monitor = threading.Thread(target=self.heartbeat, name='auth-expiry', daemon=True) self.monitor.start() # Reference disposal is best effort; immutable strings cannot be securely wiped. def heartbeat(self): while not self.closed.wait(0.1): with self.auth_lock: if time.monotonic() >= self.deadline: self.revoked.set() def authenticate(self): self.revoked.set() self.cookies.clear() password = challenge = None try: password = getpass.getpass('AlphaBot gateway password: ') challenge = hashlib.sha256(password.encode('utf-8')).hexdigest() password = None started = time.monotonic() result = self.call('/api/v1/auth/gate', {'gate_challenge': challenge}, auth=False) if (result.get('status') != 'AUTHENTICATED' or type(result.get('session_token_expires_sec')) is not int or result['session_token_expires_sec'] != 3600): raise AuthenticationFailed() # The specified JSON contains no token: require an authenticated cookie. usable = any(c.secure and c.domain.lstrip('.') == self.hostname and c.path == '/' and c.has_nonstandard_attr('HttpOnly') and (c.expires is None or c.expires > time.time()) for c in self.cookies) if not usable: raise AuthenticationFailed() with self.auth_lock: self.deadline = started + 3600 # Network time never extends the session. self.revoked.clear() except BaseException: self.cookies.clear() self.revoked.set() raise AuthenticationFailed() from None finally: password = challenge = None def ensure_auth(self): # Called in the main thread; the heartbeat never reads the terminal. if self.revoked.is_set() or time.monotonic() >= self.deadline: self.authenticate() if self.revoked.is_set() or time.monotonic() >= self.deadline: raise AuthenticationFailed() def call(self, path, body, auth=True): if auth: self.ensure_auth() request = urllib.request.Request(self.base + path, data=json.dumps(body, default=str).encode(), headers={'Content-Type': 'application/json', 'Accept': 'application/json'}, method='POST') if auth and (self.revoked.is_set() or time.monotonic() >= self.deadline): raise AuthenticationFailed() with self.http.open(request, timeout=15) as response: raw = response.read(2 * 1024 * 1024 + 1) if len(raw) > 2 * 1024 * 1024: raise Fault('oversized gateway response') result = json.loads(raw) if not isinstance(result, dict): raise Fault('invalid gateway response') return result def snapshot(self, max_age): row = self.call('/v1/snapshot', {'broker': self.broker}) age = time.time() - stamp(row['at']).timestamp() if age < -5 or age > max_age: raise Fault('stale account snapshot') if row.get('account_id') != self.account_id or type(row.get('flat')) is not bool: raise Fault('account identity/position status mismatch') if row.get('currency') != 'USD' or row.get('long_only') is not True: raise Fault('USD long-only account required') for field in ('cash', 'equity', 'asset_cost', 'exit_fees'): row[field] = number(row[field]) if row[field] < 0: raise Fault('negative account field') if type(row.get('open_orders')) is not int or row['open_orders'] < 0: raise Fault('invalid open order count') positions = row.get('positions') if not isinstance(positions, list): raise Fault('instrumentation requires complete gateway positions') seen, cost, market = set(), D('0'), D('0') for item in positions: symbol = item['symbol'] if not isinstance(symbol, str) or not symbol or symbol in seen: raise Fault('duplicate/invalid position symbol') seen.add(symbol) for field in ('quantity', 'price', 'cost'): item[field] = number(item[field]) if item['quantity'] <= 0 or item['price'] <= 0 or item['cost'] < 0: raise Fault('invalid position metrics') cost += item['cost'] market += item['quantity'] * item['price'] if cost != row['asset_cost'] or row['cash'] + market != row['equity']: raise Fault('position totals do not reconcile exactly to snapshot') if row['flat'] != (not positions): raise Fault('flat flag contradicts position snapshot') return row def close(self): self.revoked.set() self.closed.set() monitor = getattr(self, 'monitor', None) if monitor: monitor.join(timeout=1) self.cookies.clear() def route(broker, symbol, quantity, client_id): """Native order mapping; the gateway signs, rounds, validates and reconciles.""" if broker not in ('alpaca', 'kraken', 'bitstamp'): raise Fault('unsupported broker') if broker == 'alpaca': if not re.fullmatch(r'[A-Z][A-Z0-9.\-]{0,14}', symbol): raise Fault('Alpaca route is reserved for equities') return {'method': 'POST', 'path': '/v2/orders', 'body': { 'symbol': symbol, 'qty': str(quantity), 'side': 'buy', 'type': 'market', 'time_in_force': 'day', 'client_order_id': client_id}} if not re.fullmatch(r'[A-Z0-9]{2,12}/USD', symbol): raise Fault('crypto route requires BASE/USD symbol') if broker == 'kraken': return {'method': 'POST', 'path': '/0/private/AddOrder', 'body': { 'pair': symbol, 'type': 'buy', 'ordertype': 'market', 'volume': str(quantity), 'cl_ord_id': client_id}} return {'method': 'POST', 'path': '/api/v2/buy/market/' + symbol.replace('/', '').lower() + '/', 'body': {'amount': str(quantity), 'client_order_id': client_id}} class Engine: def __init__(self, store, cfg, live=False): self.s, self.cfg = store, cfg self.live = live self.g = None self.capital = number(cfg['starting_capital']) self.fee = number(cfg.get('fee_rate', '0.001')) self.slip = number(cfg.get('slippage_rate', '0.001')) self.symbol = cfg['index_asset'] if self.capital <= 0 or not 0 <= self.fee < 1 or not 0 <= self.slip < 1: raise Fault('invalid capital/fee/slippage configuration') fingerprint = hashlib.sha256(json.dumps(cfg, sort_keys=True).encode()).hexdigest() if store.get('config_hash', fingerprint) != fingerprint: raise Fault('configuration differs from durable run state; use a new project') store.put(config_hash=fingerprint) self.cash = number(store.get('cash', str(self.capital))) self.qty = number(store.get('qty', '0')) self.cost = number(store.get('cost', '0')) self.peak = number(store.get('peak', str(self.capital))) self.halted = store.get('halted', False) self.last_entry = store.get('last_entry', '') self.last_at = store.get('last_at', '') self.ledger = Ledger(store.db, cfg) self.positions = [] self._frame_active = False self.diagnostics_ready = not live self._allow_entry = True self._executions = [] self._reasons = [] self._action = 'HOLD' if live: self.g = Gateway(cfg) def save(self): self.s.put(cash=self.cash, qty=self.qty, cost=self.cost, peak=self.peak, halted=self.halted, last_entry=self.last_entry, last_at=self.last_at) def account(self, price): if self.g: a = self.g.snapshot(self.cfg.get('max_snapshot_age_seconds', 15)) self.cash, self.cost = a['cash'], a['asset_cost'] self.positions = a['positions'] equity, exit_fees = a['equity'], a['exit_fees'] else: equity = self.cash + self.qty * price exit_fees = self.qty * price * (self.fee + self.slip) a = {'equity': equity, 'open_orders': 0} self.peak = max(self.peak, equity) self.save() # Persist milestone before strategy/order dispatch. floor = capital_floor(self.capital, self.cost, self.peak) return equity, exit_fees, floor, a def _halt(self, reason, price): self._action = 'HALT' self._reasons.append(reason) self.halted = True self.save() # Latch halt before contacting any order endpoint. self.s.audit('SAFE_HALT', reason=reason) if self.g: # Server must cancel pending orders, reconcile, then close all long positions. identity = self.s.get('liquidation_id') or str(uuid.uuid4()) self.s.put(liquidation_id=identity) result = self.g.call('/v1/liquidate', { 'broker': self.cfg['broker'], 'idempotency_key': identity, 'reason': reason, 'cancel_open_orders': True}) verified = self.g.snapshot(self.cfg.get('max_snapshot_age_seconds', 15)) if result.get('flat') is not True or verified.get('flat') is not True or verified['open_orders']: raise Fault('liquidation pending/unconfirmed; strategy remains halted') try: self._executions.extend(self.execution_receipts(result, price)) finally: self.cash, self.cost, self.positions = verified['cash'], verified['asset_cost'], verified['positions'] elif self.qty and price > 0: fill = price * (1 - self.slip) fee = self.qty * fill * self.fee proceeds = self.qty * fill - fee self._executions.append({'id': str(uuid.uuid4()), 'side': 'SELL', 'symbol': self.symbol, 'quantity': self.qty, 'price': fill, 'fee': fee, 'reference_price': price, 'released_cost': self.cost, 'source': 'backtest_modeled_slippage'}) self._reasons.append(f'SELL net_proceeds={proceeds}; quantity={self.qty} * fill={fill} - fee={fee}; released_cost={self.cost}') self.cash += proceeds self.qty = self.cost = D('0') self.save() if not self.g and self.qty: raise Fault('missing liquidation price; position remains halted') self.s.audit('LIQUIDATION_CONFIRMED', cash=self.cash) def _tick(self, at, price, values, tone, neutral, month): equity, fees, floor, account = self.account(price) self._context.update(cash=self.cash, equity=equity, floor=floor, peak=self.peak, modeled_exit_fees=fees, fee_rate=self.fee, slippage_rate=self.slip) self.s.audit('TELEMETRY', at=at, price=price, equity=equity, cash=self.cash, asset_cost=self.cost, peak=self.peak, floor=floor, modeled_exit_fees=fees, indpro=values, tone=tone, empty_event_month=neutral) if self.halted or self.s.get('pending'): self.halt(f'durable_halt={self.halted}; unresolved_order_id={self.s.get("pending")}; either condition prohibits strategy', price) return if equity - fees < floor or (tone is not None and tone < D('-5')): self.halt(f'floor_threat: equity={equity} - exit_drag={fees} = {equity-fees} < floor={floor}' if equity - fees < floor else f'geopolitical_tone: {tone} < -5.0', price) return if not self._allow_entry: self._action = 'SKIP' self._reasons.append('Heartbeat monitoring slice: entry disabled; account and geopolitical risk checked') return if self.live: self.assert_portal() if self.live and not self.diagnostics_ready: raise Fault('read-only diagnostics must be prepared before live strategy calls') if self.g: self.g.ensure_auth() # No strategy callback after expiry without a fresh challenge. action = decide({'cash': self.cash, 'indpro': values, 'tone': tone, 'month': month, 'last_entry_month': self.last_entry, 'lesson_count': self.ledger.count(), 'risk_multiplier': self.ledger.multiplier()}) self._reasons.append(action['reason']) if action['action'] != 'buy' or account['open_orders']: self._action = 'SKIP' if account['open_orders']: self._reasons.append(f'open_orders={account["open_orders"]} > 0; no overlapping entry') return # Re-evaluate account immediately before dispatch; gateway must enforce atomically too. equity, fees, floor, account = self.account(price) if equity - fees < floor: self.halt(f'pre_order_floor_threat: {equity} - {fees} < {floor}', price) return if account['open_orders']: self._action = 'SKIP' self._reasons.append(f'pre_order open_orders={account["open_orders"]} > 0') return budget = min(number(action['budget']), self.cash * D('.20') * self.ledger.multiplier()) fill = price * (1 + self.slip) qty = (budget / (fill * (1 + self.fee))).quantize(D('.00000001'), rounding='ROUND_DOWN') spend = qty * fill + qty * fill * self.fee if qty <= 0: self._action = 'SKIP' self._reasons.append(f'rounded_quantity={qty} <= 0; budget={budget}') return projected = equity - (spend - qty * price) projected_floor = capital_floor(self.capital, self.cost + spend, max(self.peak, projected)) if projected - fees - qty * price * (self.fee + self.slip) < projected_floor: self._action = 'SKIP' self._reasons.append(f'projected_net={projected-fees-qty*price*(self.fee+self.slip)} < projected_floor={projected_floor}; budget={budget}') self.s.audit('BUY_BLOCKED', reason=self._reasons[-1]) return identity = str(uuid.uuid4()) self.s.put(pending=identity) if self.g: current = self.view(price) intent = self.ledger.record(at.isoformat(), self.symbol, price, current, current, 'BUY', self._reasons + ['Durable ORDER_INTENT; not a confirmed fill'], self._context, kind='ORDER_INTENT') emit('FRAME', **intent['body'], frame_id=intent['frame_id'], hash=intent['hash']) result = self.g.call('/v1/order', { 'broker': self.cfg['broker'], 'idempotency_key': identity, 'native_order': route(self.cfg['broker'], self.symbol, qty, identity), 'max_total_debit': str(budget), 'starting_capital': str(self.capital), 'peak_equity': str(self.peak), 'floor': str(floor), 'require_atomic_floor_check': True}) if result.get('status') != 'filled' or result.get('idempotency_key') != identity: raise Fault('order not fully reconciled; halt required') # Independently fetch authoritative totals; never model live fills as completed. receipts = self.execution_receipts(result, price) self._executions.extend(receipts) self.account(price) if not receipts or any(r['side'] != 'BUY' or r['symbol'] != self.symbol for r in receipts): raise Fault('missing or contradictory entry fill receipts') if sum((r['quantity'] * r['price'] + r['fee'] for r in receipts), D('0')) > budget: raise Fault('authoritative fill debit exceeded budget cap') else: self._executions.append({'id': identity, 'side': 'BUY', 'symbol': self.symbol, 'quantity': qty, 'price': fill, 'fee': qty * fill * self.fee, 'reference_price': price, 'source': 'backtest_modeled_slippage'}) self.cash -= spend self.qty += qty self.cost += spend self._action = 'BUY' self._reasons.append(f'BUY quantity={qty}; quote={price}; modeled_fill={fill}; fee={qty*fill*self.fee}; total_debit={spend}; budget_cap={budget}') self.last_entry = month self.save() self.s.put(pending=None) self.s.audit('BUY_CONFIRMED', id=identity, symbol=self.symbol, modeled_quantity=qty) def view(self, price): positions = self.positions if self.g else ([{'symbol': self.symbol, 'quantity': self.qty, 'price': price, 'cost': self.cost}] if self.qty else []) equity = self.cash + sum((dec(p['quantity']) * dec(p['price']) for p in positions), D('0')) return {'cash': self.cash, 'equity': equity, 'asset_cost': self.cost, 'floor': capital_floor(self.capital, self.cost, self.peak), 'peak': self.peak, 'halted': self.halted, 'positions': positions} def assert_portal(self): port = int(self.cfg.get('ui_port', 8080)) expected = hashlib.sha256(str(Path(self.s.db.execute('PRAGMA database_list').fetchone()[2])).encode()).hexdigest() opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect()) try: with opener.open(f'http://127.0.0.1:{port}/api/health', timeout=2) as response: status = json.loads(response.read(16384)) except (OSError, ValueError, urllib.error.URLError): raise Fault('connected read-only dashboard required before live strategy') from None if (status.get('database_key') != expected or status.get('mode') != 'live' or status.get('read_only') is not True or status.get('active_clients', 0) < 1 or not status.get('ready')): raise Fault('live portal is not connected to this account database') def prepare_diagnostics(self, root): # Local read-only viewer validation is mandatory before any new live entry. from audit_viewer import open_readonly, verify, report_html if self.live: self.assert_portal() db = open_readonly(root / 'live.sqlite3') try: frame_id, at, raw = db.execute('SELECT id,at,body FROM frames ORDER BY id DESC LIMIT 1').fetchone() meta = {'as_of_frame': frame_id, 'since': 'first instrumented snapshot', 'until': at, 'snapshot': json.loads(raw), 'integrity': verify(db, frame_id)} report = report_html(db, meta, 'id<=?', (frame_id,), 1000) destination = root / f'live_diagnostics_{frame_id}.html' with destination.open('x', encoding='utf-8') as output: output.write(report) self.s.audit('DIAGNOSTICS_READY', report=str(destination), frame_id=frame_id, sha256=hashlib.sha256(report.encode()).hexdigest()) self.diagnostics_ready = True finally: db.close() def execution_receipts(self, result, reference): receipts = result.get('fills') if not isinstance(receipts, list): raise Fault('instrumentation requires complete authoritative fill receipts') normalized = [] marks = {p['symbol']: dec(p['price']) for p in self.positions} for row in receipts: item = {'id': row['id'], 'side': row['side'], 'symbol': row['symbol'], 'quantity': number(row['quantity']), 'price': number(row['price']), 'fee': number(row['fee']), 'source': 'gateway_authoritative_receipt', 'reference_price': marks.get(row['symbol'], reference)} if not isinstance(item['id'], str) or not item['id']: raise Fault('invalid fill identity') if item['side'] == 'SELL': item['released_cost'] = number(row['released_cost']) normalized.append(item) return normalized def tick(self, at, price, values, tone, neutral, month, allow_entry=True, kind='PRICE'): return self.instrument(at, price, values, tone, neutral, month, lambda: self._tick(at, price, values, tone, neutral, month), allow_entry, kind) def halt(self, reason, price): if self._frame_active: return self._halt(reason, price) at = datetime.now(timezone.utc) if self.live else stamp(self.last_at or '1970-01-01') return self.instrument(at, price, [], None, False, '', lambda: self._halt(reason, price), False, 'HALT') def instrument(self, at, price, values, tone, neutral, month, operation, allow_entry, kind): if self._frame_active: raise Fault('nested instrumentation frame') self._frame_active, self._allow_entry = True, allow_entry self._executions, self._reasons, self._action = [], [], 'HOLD' self._context = {'indpro': values, 'tone': tone, 'empty_event_month': neutral, 'month': month, 'last_entry_month': self.last_entry, 'cash': self.cash, 'lesson_count': self.ledger.count(), 'risk_multiplier': self.ledger.multiplier(), 'target_deviation_pct': self.ledger.threshold, 'lesson_risk_decay': self.ledger.decay, 'entry_enabled': allow_entry, 'data_label': self.cfg.get('data_label', 'LOCAL_PROJECT_DATA_NOT_INDEPENDENTLY_VERIFIED'), 'reference_price_available': price > 0} state = (self.cash, self.qty, self.cost, self.last_entry, self.last_at) if not self.live: self.s.db.execute('BEGIN IMMEDIATE') before = self.view(price) error, result = None, None try: # The observed pre-decision account and complete positions are part of the frame. self.account(price) before = self.view(price) self._context['cash'] = self.cash operation() except BaseException as exc: error = exc self.halted, self._action = True, 'HALT' self._reasons.append(f'exception_path={type(exc).__name__}; execution outcome requires reconciliation') try: failures = ([{'code': 'EXECUTION_PATH_FAILURE', 'stage': kind, 'actual': type(error).__name__, 'details': 'Sanitized failure class; no secret-bearing exception text'}] if error else []) result = self.ledger.record(at.isoformat(), self.symbol, price, before, self.view(price), self._action, self._reasons, self._context, self._executions, kind, failures) if result['unresolved_reconciliation']: self.halted = True error = error or Fault('unclassified account movement; reconciliation required') if kind == 'PRICE': self.last_at = at.isoformat() self.save() if not self.live: self.s.db.commit() emit('FRAME', **result['body'], frame_id=result['frame_id'], hash=result['hash']) for item in result['body']['lessons']: emit('LESSON', **item, frame_id=result['frame_id']) except BaseException: if not self.live: self.s.db.rollback() self.cash, self.qty, self.cost, self.last_entry, self.last_at = state self.halted = True self.save() raise finally: self._frame_active = False if error: raise error return result def run(args): getcontext().prec = 50 root = Path(args.project).expanduser().resolve(strict=True) cfg = json.loads((root / 'config.json').read_text()) until = stamp(args.until) if not 1 <= args.start_year <= until.year: raise Fault('invalid start year/until range') live = args.mode == 'live' if live and not args.continuous: raise Fault('live mode requires --continuous') store = Store(root, args.mode) engine = None price = None try: store.ingest(root) engine = Engine(store, cfg, live) store.audit('START', mode=args.mode, macro_alignment='pre2000_observation_date; later_available_at_or_observation_assumed', strategy='monthly_INDPRO_vs_previous_3_months') if live: engine.tick(datetime.now(timezone.utc), D('0'), [], None, False, '', allow_entry=False, kind='PREFLIGHT') engine.prepare_diagnostics(root) equity, fees, floor, _ = engine.account(D('0')) if engine.halted or store.get('pending') or equity - fees < floor: engine.halt('startup_risk_intercept', D('0')) return 2 previous = None with (root / 'prices.jsonl').open(encoding='utf-8') as stream: while not STOP: if live and datetime.now(timezone.utc) >= until: break position = stream.tell() line = stream.readline(4 * 1024 * 1024 + 1) if len(line) > 4 * 1024 * 1024: raise Fault('price record exceeds 4 MiB') if not line or not line.endswith('\n'): stream.seek(position) if args.continuous and datetime.now(timezone.utc) < until: if live: store.ingest(root) now = datetime.now(timezone.utc) values, tone, neutral, month = store.indicators(now) engine.tick(now, price or D('0'), values, tone, neutral, month, allow_entry=False, kind='HEARTBEAT') if engine.halted: break time.sleep(1) continue if line: raise Fault('incomplete final price record') break row = json.loads(line) at = stamp(row['at']) if previous is not None and at <= previous: raise Fault('prices must have strictly increasing timestamps') previous = at if at.year < args.start_year or (engine.last_at and at <= stamp(engine.last_at)): continue if at > until: break if live: age = time.time() - at.timestamp() if age > cfg.get('max_price_age_seconds', 30): store.audit('PRICE_SKIPPED', source_at=at, reason=f'quote_age_seconds={age} > max_price_age_seconds={cfg.get("max_price_age_seconds",30)}') continue # Historical replay must never dispatch live orders. if age < -5: raise Fault('future live price timestamp') if live or args.continuous: store.ingest(root) # Reload changed local macro/event sources before each new quote. price = number(row['quotes'][engine.symbol]) if price <= 0: raise Fault('price must be positive') values, tone, neutral, month = store.indicators(at) engine.tick(at, price, values, tone, neutral, month) engine.last_at = at.isoformat() engine.save() if engine.halted: break if live and price is None: engine.halt('no_fresh_price_before_stop', D('0')) elif live and not engine.halted: engine.halt('runner_stopped_or_until_reached', price) store.audit('COMPLETE', halted=engine.halted, last_at=engine.last_at) return 2 if engine.halted else 0 except BaseException as exc: # Avoid exception messages/tracebacks that could contain HTTP headers or secrets. store.audit('AUTHENTICATION_FAILED' if isinstance(exc, AuthenticationFailed) else 'ERROR', kind=type(exc).__name__) if engine is None and live: store.put(halted=True) if engine is not None and isinstance(exc, AuthenticationFailed): engine.halted = True engine.save() store.audit('LIQUIDATION_UNCONFIRMED', intervention_required=True) elif engine is not None: try: engine.halt('execution_error', price or D('0')) except BaseException: store.audit('LIQUIDATION_UNCONFIRMED', intervention_required=True) return 2 finally: if engine and engine.g: engine.g.close() store.db.close() store.lock.close() def stop(signum, frame): global STOP STOP = True class ExecutionService: """Own the actual backtest runner; status sharing never shares its SQLite connection.""" def __init__(self, args=None): self.args = args self._lock = threading.Lock() self._thread = None self._status = {'state': 'observer', 'mode': args.mode if args else None, 'exit_code': None, 'error_kind': None} def snapshot(self): with self._lock: return dict(self._status) def start(self): global STOP if self.args is None: return if self.args.mode != 'backtest': raise Fault('combined server execution is restricted to backtests') if self.args.start_year is None or self.args.until is None: raise Fault('combined execution requires start-year and until') if self._thread is not None: raise Fault('execution service already started') STOP = False with self._lock: self._status['state'] = 'running' self._thread = threading.Thread(target=self._execute, name='alphabot-backtest', daemon=False) self._thread.start() def _execute(self): getcontext().prec = 50 try: code = run(self.args) # The real ingestion, decide(), floor and journal path. with self._lock: self._status.update(state='completed' if code == 0 else 'halted', exit_code=code) except BaseException as error: with self._lock: self._status.update(state='failed', exit_code=2, error_kind=type(error).__name__) def stop(self): global STOP if self._thread is None: return if self._thread.is_alive(): STOP = True self._thread.join(timeout=5) if self._thread.is_alive(): with self._lock: self._status['state'] = 'stopping' class DashboardDatabase: """Connections belong to individual worker calls; never share a SQLite cursor.""" def __init__(self, project, mode): self.path = Path(project).expanduser().resolve(strict=True) / f'{mode}.sqlite3' self.key = hashlib.sha256(str(self.path).encode()).hexdigest() self.mode = mode def read(self, kind, query): getcontext().prec = 50 # Each executor thread owns its Decimal context. from audit_viewer import open_readonly, timestamp deadline = time.monotonic() + 5 def check(): if time.monotonic() > deadline: raise TimeoutError('read budget exceeded') if not self.path.exists(): return {'ready': False, 'rows': [], 'last_id': 0} db = open_readonly(self.path) db.set_progress_handler(lambda: int(time.monotonic() > deadline), 10000) try: try: first, last, count, last_id = db.execute( 'SELECT min(at),max(at),count(*),coalesce(max(id),0) FROM frames').fetchone() except sqlite3.OperationalError: return {'ready': False, 'rows': [], 'last_id': 0} bounds = {'first': first, 'last': last, 'count': count, 'last_id': last_id} if kind == 'health': return {'ready': bool(count), 'database_key': self.key, 'mode': self.mode, 'read_only': True, 'bounds': bounds} if kind == 'stream': cursor = int(query.get('after', '0')) if cursor < 0: raise ValueError('invalid stream cursor') rows, size = [], 0 for fid, raw, checksum in db.execute( 'SELECT id,body,hash FROM frames WHERE id>? ORDER BY id LIMIT 100', (cursor,)): check() if len(raw.encode()) > 512000: raise ValueError('frame exceeds bounded stream size; use local JSON viewer') if rows and size + len(raw.encode()) > 512000: break rows.append({'id': fid, 'body': json.loads(raw), 'hash': checksum}) size += len(raw.encode()) return {'ready': bool(count), 'rows': rows, 'last_id': last_id} if not count: return {'ready': False, 'bounds': bounds, 'points': [], 'frames': [], 'lessons': []} start = timestamp(query['from']) if query.get('from') else first end = timestamp(query['until'], True) if query.get('until') else last if start > end: raise ValueError('window start exceeds end') cursor = int(query.get('cursor', '0')) if cursor < 0: raise ValueError('invalid page cursor') selected = db.execute('SELECT id,body FROM frames WHERE at<=? ORDER BY at DESC,id DESC LIMIT 1', (end,)).fetchone() if kind == 'lessons': counts, rows, more = {}, [], False for fid, seq, raw, at, symbol in db.execute( "SELECT l.frame_id,l.seq,l.body,f.at,json_extract(f.body,'$.symbol') " 'FROM lessons l JOIN frames f ON f.id=l.frame_id WHERE f.at<=? ORDER BY l.frame_id,l.seq', (end,)): check() item = json.loads(raw) details = item.get('details') if isinstance(item.get('details'), dict) else {} signature = (item['code'], item['stage'], details.get('side'), details.get('source'), symbol) counts[signature] = counts.get(signature, 0) + 1 if at < start or fid <= cursor: continue if len(rows) >= 100 and fid != rows[-1]['frame_id']: more = True break solution = {'TARGET_DEVIATION': 'Review reference quote and execution costs; retain the recorded risk decay.', 'UNCLASSIFIED_MOVEMENT': 'Reconcile cash and authoritative fill receipts; keep the halt latched.', 'EXECUTION_PATH_FAILURE': 'Resolve the recorded failure stage, rerun checks, and obtain administrator review.'}.get(item['code'], 'Administrator review required; no automatic remediation.') if len(rows) >= 1000: raise ValueError('lesson group exceeds bounded page size') rows.append({'id': f'{fid}.{seq}', 'frame_id': fid, 'at': at, 'record': item, 'magnitude_pct': details.get('deviation_pct'), 'occurrence': counts[signature], 'REPEATED_DISCREPANCY': counts[signature] > 1, 'solution': solution, 'solution_status': 'PROPOSED_NOT_EXECUTED'}) return {'ready': True, 'lessons': rows, 'next_cursor': rows[-1]['frame_id'] if more and rows else None} total = db.execute('SELECT count(*) FROM frames WHERE at>=? AND at<=?', (start, end)).fetchone()[0] limit = max(2, min(1000, int(query.get('points', '1000')))) stride = max(1, (total + limit - 1) // limit) points = [] for index, (fid, raw) in enumerate(db.execute('SELECT id,body FROM frames WHERE at>=? AND at<=? ORDER BY at,id', (start, end))): check() if index % stride and index != total - 1: continue b = json.loads(raw) points.append({'id': fid, 'at': b['at'], 'price': b['price'], 'price_available': b['context'].get('reference_price_available', True), 'equity': b['after']['equity'], 'cash': b['after']['cash'], 'floor': b['after']['floor'], 'risk': b['risk_multiplier_after'], 'lessons': b['lesson_count'], 'action': b['action']}) rows, page_size, more = [], 0, False for fid, raw in db.execute('SELECT id,body FROM frames WHERE id>? AND at>=? AND at<=? ORDER BY id LIMIT 101', (cursor, start, end)): check() size = len(raw.encode()) if size > 512000: raise ValueError('oversized frame; use complete local viewer') if rows and (len(rows) >= 100 or page_size + size > 512000): more = True break rows.append({'id': fid, 'body': json.loads(raw)}) page_size += size distribution = [] if selected: after = json.loads(selected[1])['after'] for position in after['positions']: value = dec(position['quantity']) * dec(position['price']) weight = value / dec(after['equity']) * 100 if dec(after['equity']) else D('0') distribution.append({**position, 'market_value': format(value,'f'), 'weight_pct': format(weight,'f')}) return {'ready': True, 'distribution': distribution, 'bounds': bounds, 'window': {'from': start, 'until': end, 'count': total}, 'snapshot': {'id': selected[0], 'body': json.loads(selected[1])} if selected else None, 'points': points, 'frames': rows[:100], 'sampled': total > len(points), 'next_cursor': rows[-1]['id'] if more else None} finally: db.close() async def dashboard_send(ws, payload): """One bounded write in flight per client; a failed socket never escapes the handler.""" try: text = json.dumps(fixed(payload), allow_nan=False) if len(text.encode()) > 1048576: return False await asyncio.wait_for(ws.send_str(text), timeout=2) return True except (OSError, RuntimeError, asyncio.TimeoutError): return False def create_dashboard(project, mode='backtest', port=8080, execute_args=None): from aiohttp import web source = DashboardDatabase(project, mode) service = ExecutionService(execute_args) assets = Path(__file__).parent / 'web' clients = set() pool = ThreadPoolExecutor(max_workers=2, thread_name_prefix='audit-reader') jobs = asyncio.Semaphore(2) origins = {f'http://127.0.0.1:{port}', f'http://localhost:{port}'} hosts = {f'127.0.0.1:{port}', f'localhost:{port}'} async def query(kind, values): async with jobs: return await asyncio.get_running_loop().run_in_executor(pool, source.read, kind, dict(values)) active_requests = 0 @web.middleware async def boundary(request, handler): nonlocal active_requests if active_requests >= 32: raise web.HTTPServiceUnavailable(text='Read capacity reached') active_requests += 1 try: if request.host not in hosts or (request.headers.get('Origin') and request.headers['Origin'] not in origins): raise web.HTTPForbidden(text='Loopback same-origin access required') if request.method != 'GET': raise web.HTTPMethodNotAllowed(request.method, ['GET']) try: response = await handler(request) except (ValueError, KeyError): response = web.json_response({'error': 'INVALID_WINDOW_OR_CURSOR'}, status=400) except (sqlite3.Error, OSError, TimeoutError): response = web.json_response({'error': 'DATABASE_BUSY_OR_UNAVAILABLE'}, status=503) response.headers['Cache-Control'] = 'no-store' response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['Referrer-Policy'] = 'no-referrer' response.headers['X-Frame-Options'] = 'DENY' return response finally: active_requests -= 1 app = web.Application(middlewares=[boundary], client_max_size=1024, handler_args={'timeout_ceil_threshold': 1}) async def static(request): name = request.match_info.get('name') or 'index.html' if name not in ('index.html', 'monitor.html', 'styles.css', 'demo.json'): raise web.HTTPNotFound() content = (assets / name).read_text() hashes = ['sha256-' + base64.b64encode(hashlib.sha256(block.encode()).digest()).decode() for block in re.findall(r'', content, re.S)] policy = "default-src 'none'; style-src 'self'; script-src " + (' '.join("'"+x+"'" for x in hashes) or "'none'") policy += "; connect-src 'self' ws://127.0.0.1:"+str(port)+" ws://localhost:"+str(port)+"; base-uri 'none'; form-action 'none'; frame-ancestors 'none'" return web.Response(text=content, content_type='application/json' if name.endswith('.json') else ('text/css' if name.endswith('.css') else 'text/html'), headers={'Content-Security-Policy': policy}) async def api(request): kind = request.match_info['kind'] result = await query(kind, request.query) result['execution'] = service.snapshot() if kind == 'health': result.update(active_clients=len(clients), database_key=source.key, mode=mode, read_only=True) encoded = json.dumps(fixed(result), allow_nan=False) if len(encoded.encode()) > 2097152: return web.json_response({'error': 'WINDOW_TOO_LARGE_USE_PAGINATION'}, status=413) return web.Response(text=encoded, content_type='application/json') async def stream(request): if request.headers.get('Origin') not in origins: raise web.HTTPForbidden() if len(clients) >= 8: raise web.HTTPServiceUnavailable(text='Client capacity reached') cursor = int(request.query.get('after', '0')) if cursor < 0: raise ValueError('negative cursor') ws = web.WebSocketResponse(heartbeat=15, max_msg_size=1024, writer_limit=65536, compress=False) clients.add(ws) # Reserve capacity before the upgrade awaits. receiver = None try: await ws.prepare(request) if request.transport: request.transport.set_write_buffer_limits(high=65536, low=16384) receiver = asyncio.create_task(ws.receive()) while not ws.closed: batch = await query('stream', {'after': cursor}) if not await dashboard_send(ws, {'type': 'transactions', **batch, 'execution': service.snapshot()}): break if batch['rows']: cursor = batch['rows'][-1]['id'] done, _ = await asyncio.wait({receiver}, timeout=0.5) if done: break # Client commands are unsupported; this stream is read-only. except (OSError, RuntimeError, ValueError, sqlite3.Error, asyncio.TimeoutError): pass finally: clients.discard(ws) if receiver: receiver.cancel() await asyncio.gather(receiver, return_exceptions=True) try: await asyncio.wait_for(ws.close(), 1) except (OSError, RuntimeError, asyncio.TimeoutError): if request.transport: request.transport.abort() return ws async def shutdown(app): try: await asyncio.wait_for(asyncio.gather(*(ws.close(code=1001) for ws in tuple(clients)), return_exceptions=True), 2) except asyncio.TimeoutError: pass pool.shutdown(wait=False, cancel_futures=True) async def execution_lifecycle(app): service.start() try: yield finally: await asyncio.to_thread(service.stop) app.cleanup_ctx.append(execution_lifecycle) app.router.add_get('/ws', stream) app.router.add_get('/api/{kind:health|window|lessons}', api) app.router.add_get('/', static) app.router.add_get('/{name:index.html|monitor.html|styles.css|demo.json}', static) app.on_shutdown.append(shutdown) return app def serve_dashboard(args): if not 1 <= args.port <= 65535: raise Fault('port must be between 1 and 65535') from aiohttp import web web.run_app(create_dashboard(args.project, args.mode, args.port, execute_args=args if args.execute else None), host='127.0.0.1', port=args.port, access_log=None, print=None, shutdown_timeout=3) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--project', required=True) parser.add_argument('--start-year', type=int) parser.add_argument('--until') parser.add_argument('--serve', action='store_true') parser.add_argument('--execute', action='store_true', help='run the actual backtest alongside --serve; dates also imply this mode') parser.add_argument('--port', type=int, default=8080) parser.add_argument('--continuous', action='store_true') parser.add_argument('--mode', choices=('backtest', 'live'), default='backtest') args = parser.parse_args() if args.serve: args.execute = args.execute or args.start_year is not None or args.until is not None if args.execute and (args.start_year is None or args.until is None): parser.error('combined execution requires --start-year and --until') if args.execute and args.mode != 'backtest': parser.error('combined execution is backtest-only; --serve alone is an observer') serve_dashboard(args) return 0 if args.execute: parser.error('--execute requires --serve') if args.start_year is None or args.until is None: parser.error('--start-year and --until are required for execution') signal.signal(signal.SIGTERM, stop) signal.signal(signal.SIGINT, stop) try: return run(args) except (OSError, ValueError, KeyError, Fault): emit('STARTUP_REJECTED', reason='invalid project, configuration, or input') return 2 if __name__ == '__main__': raise SystemExit(main())