''' + 'AlphaBot · audit snapshot
Read-only · local · USD · fixed-point Decimal
'+esc(meta['snapshot']['context'].get('data_label','Unlabeled historical record'))+'
'+\
'Recorded as of '+esc(meta['snapshot']['at'])+'; requested window '+esc(meta['since'])+' to '+esc(meta['until'])+'
'+\
''+''.join('
'+esc(k)+''+esc(after[k])+'
' for k in ('equity','cash','floor','peak'))+'
Observed data, not a regulatory certification. Live readiness is unverified. Hash chain verified for '+esc(meta['integrity']['verified_frames'])+' frames; no external anchor.
'+\
'Performance and protection
Equity: green · Floor: amber · Cash: blue. '+esc(count)+' recorded window frames; '+esc(len(points))+' plotted points, downsampled by sequence with endpoints retained.
'+chart+\
'All positions at snapshot
Cash share: '+esc(format(dec(after['cash'])/equity*100 if equity else D(0),'f'))+'%.
'+\
'Trial balance through selected frame
'+\
'Decision frames
Showing the latest '+esc(len(recent))+' of '+esc(count)+' window frames. JSON export includes every selected frame and all postings.
'+\
'LESSON paths in displayed frames
'+(''.join(lesson_rows) or 'No lessons in displayed frames.
')+\
'Selected decision context
'+esc(json.dumps(meta['snapshot']['context'],indent=2))+'
'
def main():
p=argparse.ArgumentParser(description=__doc__)
p.add_argument('--project',required=True)
p.add_argument('--mode',choices=('backtest','live'),default='backtest')
group=p.add_mutually_exclusive_group()
group.add_argument('--date');group.add_argument('--milestone')
p.add_argument('--from',dest='since');p.add_argument('--until')
p.add_argument('--format',choices=('json','html'),default='json')
p.add_argument('--output');p.add_argument('--max-points',type=int,default=1000)
args=p.parse_args()
if args.max_points<2:p.error('--max-points must be at least 2')
if (args.date or args.milestone) and args.until:p.error('--until cannot combine with --date/--milestone')
db=None;out=None
try:
source=Path(args.project).expanduser().resolve(strict=True)/f'{args.mode}.sqlite3'
if args.output:
destination=Path(args.output).expanduser().resolve()
if destination in (source,Path(str(source)+'-wal'),Path(str(source)+'-shm')):
raise ValueError('report output cannot overwrite database or WAL files')
db=open_readonly(source)
end_id,start,end,where,params=selection(db,args)
integrity=verify(db,end_id)
meta={'as_of_frame':end_id,'since':start,'until':end[0],
'snapshot':json.loads(end[1]),'integrity':integrity}
out=open(args.output,'x',encoding='utf-8') if args.output else sys.stdout
if args.format=='html':out.write(report_html(db,meta,where,params,args.max_points))
else:
out.write('{"metadata":'+json.dumps(meta)+',"frames":[')
first=True
for fid,raw in db.execute('SELECT id,body FROM frames WHERE '+where+' ORDER BY at,id',params):
if not first:out.write(',')
out.write('{"frame_id":'+str(fid)+',"body":'+raw+'}');first=False
out.write(']}\n')
return 0
except (OSError,ValueError,KeyError,TypeError,sqlite3.Error) as exc:
print(json.dumps({'event':'VIEW_REJECTED','kind':type(exc).__name__}),file=sys.stderr)
return 2
finally:
if out is not None and out is not sys.stdout:out.close()
if db:db.close()
if __name__=='__main__':
raise SystemExit(main())
```
## scripts/gauntlet_v2.py
```python
#!/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
from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
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):
# 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)))
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 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 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
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):
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')
store.ingest(root) # Refresh local macro/event files at each live tick.
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
def main():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument('--project', required=True)
parser.add_argument('--start-year', required=True, type=int)
parser.add_argument('--until', required=True)
parser.add_argument('--continuous', action='store_true')
parser.add_argument('--mode', choices=('backtest', 'live'), default='backtest')
args = parser.parse_args()
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())
```
## scripts/host_strategy.py
```python
# SPDX-License-Identifier: MIT
"""AlphaBot V2 declarative monthly momentum; all economics use Decimal."""
from decimal import Decimal as D
def capital_floor(starting_capital, asset_cost, peak_equity):
minimum, milestone = D('250'), D('1000')
while peak_equity >= milestone:
minimum, milestone = milestone / 2, milestone * 2
return max(minimum, starting_capital - asset_cost)
def decide(context):
tone = context['tone']
if tone is not None and tone < D('-5'):
return {'action': 'liquidate', 'reason': f'GDELT tone={tone} < -5.0'}
if tone is None:
return {'action': 'hold', 'reason': 'GDELT tone unknown: recorded events are not yet contemporaneously visible'}
values = context['indpro']
if len(values) != 4:
return {'action': 'hold', 'reason': f'visible consecutive INDPRO months={len(values)} != 4'}
if context['month'] == context['last_entry_month']:
return {'action': 'hold', 'reason': f'observation month={context["month"]} equals last_entry_month; one-entry limit'}
average = sum(values[:-1]) / 3
multiplier = context.get('risk_multiplier', D('1'))
if values[-1] > average:
budget = context['cash'] * D('.20') * multiplier
return {'action': 'buy', 'budget': budget,
'reason': f'INDPRO_current={values[-1]} > prior_three_mean={average}; cash={context["cash"]} * 0.20 * risk_multiplier={multiplier} = budget={budget}; prior_lessons={context.get("lesson_count",0)}'}
return {'action': 'hold', 'reason': f'INDPRO_current={values[-1]} <= prior_three_mean={average}'}
```
## tests/test_core.py
```python
# SPDX-License-Identifier: MIT
import contextlib
from datetime import datetime, timezone
from decimal import Decimal as D
import io
import json
from pathlib import Path
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'scripts'))
import gauntlet_v2 as g
from host_strategy import capital_floor, decide
class CoreTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.root = Path(self.tmp.name)
self.store = g.Store(self.root, 'backtest')
self.cfg = {'starting_capital': '500', 'index_asset': 'BTC/USD', 'fee_rate': '0.001', 'slippage_rate': '0.001'}
self.out = io.StringIO()
self.redirect = contextlib.redirect_stdout(self.out)
self.redirect.__enter__()
def tearDown(self):
self.store.db.close()
self.store.lock.close()
self.redirect.__exit__(None,None,None)
self.tmp.cleanup()
def write(self, name, rows):
(self.root/name).write_text(''.join(json.dumps(r)+'\n' for r in rows))
def engine(self):
return g.Engine(self.store, self.cfg)
def tick(self,e,price='100',tone='0',month='1999-04'):
e.tick(g.stamp('1999-04-01'),D(price),[D(1),D(2),D(3),D(4)],D(tone),True,month)
def test_floor_exact_milestones(self):
for peak, expected in [('999.99','250'),('1000','500'),('1999.99','500'),('2000','1000'),('4000','2000'),('8000','4000')]:
self.assertEqual(capital_floor(D(500),D(490),D(peak)),D(expected))
def test_cash_formula(self):
self.assertEqual(capital_floor(D(500),D('100.10'),D(500)),D('399.90'))
def test_initial_equality_permits_entry(self):
e=self.engine();self.tick(e)
self.assertGreater(e.qty,0);self.assertFalse(e.halted)
def test_twenty_percent_includes_fees(self):
e=self.engine();self.tick(e)
self.assertLessEqual(e.cost,D(100));self.assertGreater(e.cost,D('99.99'))
self.assertEqual(e.cash+e.cost,D(500))
def test_once_monthly(self):
e=self.engine();self.tick(e);q=e.qty;self.tick(e)
self.assertEqual(e.qty,q)
def test_next_month_entry(self):
e=self.engine();self.tick(e);q=e.qty;self.tick(e,month='1999-05')
self.assertGreater(e.qty,q)
def test_floor_loss_halts_and_liquidates(self):
e=self.engine()
for m in range(1,5):self.tick(e,month=f'1999-{m:02}')
self.tick(e,price='1')
self.assertTrue(e.halted);self.assertEqual(e.qty,0)
def test_tone_exit(self):
e=self.engine();self.tick(e);self.tick(e,tone='-5.01')
self.assertTrue(e.halted);self.assertEqual(e.qty,0)
def test_tone_threshold_is_strict(self):
e=self.engine();self.tick(e,tone='-5');self.assertFalse(e.halted)
def test_restart_preserves_peak(self):
e=self.engine();e.peak=D(4000);e.save()
self.assertEqual(self.engine().peak,D(4000))
def test_restart_halt_latched(self):
e=self.engine();e.halt('test',D(100));e=self.engine();self.tick(e)
self.assertTrue(e.halted);self.assertEqual(e.qty,0)
def test_pending_order_halts(self):
e=self.engine();self.store.put(pending='unknown');self.tick(e)
self.assertTrue(e.halted);self.assertEqual(e.qty,0)
def test_config_change_rejected(self):
self.engine();self.cfg['starting_capital']='600'
with self.assertRaises(g.Fault):self.engine()
def test_no_shorting_or_negative_cash(self):
e=self.engine()
for m in range(1,13):self.tick(e,month=f'1999-{m:02}')
self.assertGreaterEqual(e.cash,0);self.assertGreaterEqual(e.qty,0)
def test_nan_rejected(self):
for v in ['NaN','Infinity','-Infinity','bad']:
with self.assertRaises(g.Fault):g.number(v)
def test_naive_datetime_rejected(self):
with self.assertRaises(g.Fault):g.stamp('2026-01-01T12:00:00')
def test_missing_geo_rejected(self):
self.write('macro.jsonl',[])
with self.assertRaises(g.Fault):self.store.ingest(self.root)
def test_pre2000_observation_override_and_no_future_geo(self):
self.write('macro.jsonl',[{'series':'INDPRO','observation_date':f'1999-{m:02}-01','value':str(m),'available_at':'2001-01-01'} for m in range(1,5)])
self.write('geopolitical.jsonl',[{'at':'1999-04-20','tone':'-9'}])
self.store.ingest(self.root)
vals,tone,neutral,month=self.store.indicators(g.stamp('1999-04-01'))
self.assertEqual(vals,[D(1),D(2),D(3),D(4)]);self.assertIsNone(tone);self.assertFalse(neutral)
self.assertEqual(self.store.indicators(g.stamp('1999-04-21'))[1],D(-9))
def test_post2000_vintage_hidden(self):
self.write('macro.jsonl',[{'series':'INDPRO','observation_date':'2001-01-01','value':'1','available_at':'2001-03-01'}])
self.write('geopolitical.jsonl',[]);self.store.ingest(self.root)
vals,tone,neutral,month=self.store.indicators(g.stamp('2001-02-01'))
self.assertEqual(month,'');self.assertEqual(vals,[]);self.assertEqual(tone,0);self.assertTrue(neutral)
def test_missing_month_blocks_signal(self):
self.write('macro.jsonl',[{'series':'INDPRO','observation_date':f'1999-{m:02}-01','value':str(m)} for m in (1,2,4,5)])
self.write('geopolitical.jsonl',[]);self.store.ingest(self.root)
self.assertEqual(self.store.indicators(g.stamp('1999-05-01'))[0],[])
def test_concurrent_runner_rejected(self):
with self.assertRaises(g.Fault):g.Store(self.root,'backtest')
def test_three_native_order_mappings(self):
self.assertEqual(g.route('alpaca','SPY',D(1),'abc')['path'],'/v2/orders')
self.assertEqual(g.route('kraken','BTC/USD',D(1),'abc')['body']['ordertype'],'market')
self.assertEqual(g.route('bitstamp','BTC/USD',D(1),'abc')['path'],'/api/v2/buy/market/btcusd/')
def test_wrong_asset_broker_rejected(self):
with self.assertRaises(g.Fault):g.route('alpaca','BTC/USD',D(1),'abc')
with self.assertRaises(g.Fault):g.route('kraken','SPY',D(1),'abc')
def test_unknown_tone_holds(self):
self.assertEqual(decide({'tone':None})['action'],'hold')
def test_mocked_live_order_requires_fill_reconciliation(self):
e=self.engine()
class MockGateway:
def ensure_auth(self):pass
def snapshot(self,age):return {'cash':D(500),'asset_cost':D(0),'equity':D(500),'exit_fees':D(0),'open_orders':0,'positions':[]}
def call(self,path,body):return {'status':'pending'}
self.cfg['broker']='kraken';e.g=MockGateway()
with self.assertRaises(g.Fault):self.tick(e)
self.assertIsNotNone(self.store.get('pending'))
def test_gateway_rejects_http_and_mismatched_host(self):
for url in ['http://example.com','https://wrong.example','https://example.com/path']:
with patch.dict(g.os.environ,{'LIVE_AUTH_GATEWAY_URL':url}):
with self.assertRaises(g.Fault):g.Gateway({'broker':'kraken','account_id':'a','gateway_hostname':'example.com'})
def gate(self):
obj=g.Gateway.__new__(g.Gateway)
obj.deadline=0;obj.revoked=g.threading.Event();obj.revoked.set()
obj.closed=g.threading.Event();obj.auth_lock=g.threading.Lock()
obj.cookies=g.http.cookiejar.CookieJar();obj.hostname='example.com'
return obj
def cookie(self,obj,secure=True):
obj.cookies.set_cookie(g.http.cookiejar.Cookie(0,'session','test',None,False,
'example.com',False,False,'/',True,secure,None,True,None,None,{'HttpOnly':None},False))
def test_expiry_requires_fresh_authentication(self):
obj=self.gate()
with patch.object(obj,'authenticate',side_effect=g.AuthenticationFailed()) as auth:
with self.assertRaises(g.AuthenticationFailed):obj.ensure_auth()
auth.assert_called_once()
def test_exact_gate_contract(self):
obj=self.gate();seen=[]
def call(path,body,auth=True):
seen.append((path,body,auth));self.cookie(obj)
return {'status':'AUTHENTICATED','session_token_expires_sec':3600}
obj.call=call
with patch.object(g.getpass,'getpass',return_value='test password'):
before=g.time.monotonic();obj.authenticate()
self.assertEqual(seen,[('/api/v1/auth/gate',{'gate_challenge':g.hashlib.sha256(b'test password').hexdigest()},False)])
self.assertFalse(obj.revoked.is_set());self.assertLessEqual(obj.deadline,before+3601)
def test_gate_status_must_match_exactly(self):
obj=self.gate();obj.call=lambda *a,**k:{'status':'authenticated','session_token_expires_sec':3600}
with patch.object(g.getpass,'getpass',return_value='test'):
with self.assertRaisesRegex(g.AuthenticationFailed,'AUTHENTICATION_FAILED'):obj.authenticate()
self.assertTrue(obj.revoked.is_set())
def test_gate_without_session_binding_rejected(self):
obj=self.gate();obj.call=lambda *a,**k:{'status':'AUTHENTICATED','session_token_expires_sec':3600}
with patch.object(g.getpass,'getpass',return_value='test'):
with self.assertRaises(g.AuthenticationFailed):obj.authenticate()
def test_unreachable_gate_fails_immediately(self):
obj=self.gate()
with patch.object(obj,'call',side_effect=OSError('unreachable')) as call:
with patch.object(g.getpass,'getpass',return_value='test'):
with self.assertRaises(g.AuthenticationFailed):obj.authenticate()
call.assert_called_once()
def test_wrong_expiry_rejected(self):
obj=self.gate();obj.call=lambda *a,**k:{'status':'AUTHENTICATED','session_token_expires_sec':7200}
with patch.object(g.getpass,'getpass',return_value='test'):
with self.assertRaises(g.AuthenticationFailed):obj.authenticate()
def test_heartbeat_revokes_expired_session(self):
obj=self.gate();obj.revoked.clear();obj.deadline=g.time.monotonic()-1
t=g.threading.Thread(target=obj.heartbeat);t.start()
self.assertTrue(obj.revoked.wait(1));obj.closed.set();t.join(1)
def test_current_session_does_not_reprompt(self):
obj=self.gate();obj.deadline=g.time.monotonic()+60;obj.revoked.clear()
with patch.object(obj,'authenticate') as auth:obj.ensure_auth();auth.assert_not_called()
def test_missing_liquidation_price_does_not_erase_position(self):
e=self.engine();self.tick(e)
with self.assertRaises(g.Fault):e.halt('missing_price',D(0))
self.assertGreater(e.qty,0);self.assertTrue(e.halted)
if __name__=='__main__':unittest.main(verbosity=2)
```
## tests/test_instrumentation.py
```python
# SPDX-License-Identifier: MIT
import contextlib
from decimal import Decimal as D
import hashlib
import io
import json
from pathlib import Path
import sqlite3
import subprocess
import sys
import tempfile
import unittest
from unittest.mock import patch
sys.path.insert(0,str(Path(__file__).resolve().parents[1] / 'scripts'))
import gauntlet_v2 as g
import audit_viewer as v
from audit_ledger import Ledger,canonical,digest
class InstrumentationTests(unittest.TestCase):
def setUp(self):
self.tmp=tempfile.TemporaryDirectory();self.root=Path(self.tmp.name)
self.store=g.Store(self.root,'backtest')
self.cfg={'starting_capital':'500','index_asset':'BTC/USD','slippage_rate':'0.001','fee_rate':'0.001'}
self.e=g.Engine(self.store,self.cfg)
self.redirect=contextlib.redirect_stdout(io.StringIO());self.redirect.__enter__()
def tearDown(self):
self.store.db.close();self.store.lock.close();self.redirect.__exit__(None,None,None);self.tmp.cleanup()
def tick(self,month='1999-04',price='100',tone='0',values=None,**kwargs):
return self.e.tick(g.stamp(month+'-01'),D(price),values or [D(1),D(2),D(3),D(4)],
D(tone),True,month,**kwargs)
def last(self):
return json.loads(self.store.db.execute('SELECT body FROM frames ORDER BY id DESC LIMIT 1').fetchone()[0])
def verify(self):
end=self.store.db.execute('SELECT max(id) FROM frames').fetchone()[0]
return v.verify(self.store.db,end)
def test_buy_journal_is_balanced_and_reconciles_holdings(self):
self.tick();body=self.last();checked=self.verify()
self.assertEqual(body['action'],'BUY');self.assertEqual(body['debit_total'],body['credit_total'])
self.assertEqual(D(checked['balances']['Assets:Cash']),self.e.cash)
self.assertEqual(D(checked['balances']['Assets:Inventory']),self.e.cost)
def test_fee_is_capitalized_once_and_disclosed(self):
self.tick();b=self.last();fill=b['executions'][0]
fee=D(fill['quantity'])*D(fill['price'])*D('.001')
self.assertEqual(D(b['fee_drag']),fee)
self.assertEqual(self.e.cost,D(fill['quantity'])*D(fill['price'])+fee)
self.assertNotIn('Expenses:EntryFee',self.verify()['balances'])
def test_each_hold_has_a_frame_and_reason(self):
self.tick();self.tick()
self.assertEqual(self.store.db.execute('SELECT count(*) FROM frames').fetchone()[0],2)
self.assertEqual(self.last()['action'],'SKIP');self.assertIn('one-entry limit',self.last()['reasons'][0])
self.verify()
def test_hold_marks_to_market_without_fabricated_trade(self):
self.tick();self.tick(price='110')
self.assertEqual(self.last()['executions'],[]);self.assertEqual(self.last()['fee_drag'],'0')
self.assertGreater(D(self.verify()['balances']['Assets:ValuationAdjustment']),0)
def test_sell_realized_and_unrealized_balances_reconcile(self):
self.tick();self.tick(price='110',tone='-6')
b=self.last();self.assertEqual(b['action'],'HALT');self.assertEqual(b['executions'][0]['side'],'SELL')
trial=self.verify()['balances'];self.assertEqual(D(trial['Assets:Inventory']),0)
self.assertEqual(D(trial['Assets:ValuationAdjustment']),0)
self.assertEqual(D(trial['Assets:Cash']),self.e.cash)
def test_exact_deviation_threshold_not_float_comparison(self):
self.cfg['slippage_rate']='0.000001';self.e=g.Engine.__new__(g.Engine)
# Use a separate pristine project because config fingerprint must remain immutable.
other=g.Store(self.root,'precise');e=g.Engine(other,self.cfg)
try:
e.tick(g.stamp('1999-04-01'),D(100),[D(1),D(2),D(3),D(4)],D(0),True,'1999-04')
lesson=json.loads(other.db.execute('SELECT body FROM lessons').fetchone()[0])
self.assertEqual(D(lesson['details']['deviation_pct']),D('0.0001'))
finally:other.db.close();other.lock.close()
def test_below_threshold_no_lesson(self):
self.cfg['slippage_rate']='0.0000001';other=g.Store(self.root,'precise');e=g.Engine(other,self.cfg)
try:
e.tick(g.stamp('1999-04-01'),D(100),[D(1),D(2),D(3),D(4)],D(0),True,'1999-04')
self.assertEqual(e.ledger.count(),0)
finally:other.db.close();other.lock.close()
def test_lessons_constrain_next_budget_and_survive_restart(self):
self.tick();self.assertEqual(self.e.ledger.count(),1)
cash=self.e.cash;self.tick(month='1999-05');b=self.last()
self.assertEqual(D(b['context']['risk_multiplier']),D('.95'))
self.assertLessEqual(sum(D(x['quantity'])*D(x['price'])+D(x['fee']) for x in b['executions']),cash*D('.20')*D('.95'))
reloaded=g.Engine(self.store,self.cfg);self.assertEqual(reloaded.ledger.count(),2)
self.assertEqual(reloaded.ledger.multiplier(),D('.9025'))
def test_heartbeat_does_not_place_order_or_advance_price_cursor(self):
self.tick();last=self.e.last_at
self.tick(month='1999-05',allow_entry=False,kind='HEARTBEAT')
self.assertEqual(self.last()['executions'],[]);self.assertEqual(self.e.last_at,last)
self.assertEqual(self.last()['kind'],'HEARTBEAT')
def test_live_snapshot_requires_complete_positions(self):
gate=g.Gateway.__new__(g.Gateway);gate.account_id='acct';gate.broker='kraken'
base={'at':g.datetime.now(g.timezone.utc).isoformat(),'account_id':'acct','flat':True,
'currency':'USD','long_only':True,'cash':'500','equity':'500','asset_cost':'0','exit_fees':'0','open_orders':0}
gate.call=lambda *a,**k:dict(base)
with self.assertRaises(g.Fault):gate.snapshot(15)
base['positions']=[];self.assertEqual(gate.snapshot(15)['positions'],[])
def test_reconciliation_suspense_is_explicit_and_halts(self):
self.tick();self.e.cash+=D(1)
with self.assertRaises(g.Fault):self.tick()
self.assertTrue(self.e.halted);self.assertTrue(self.last()['unresolved_reconciliation'])
self.assertEqual(self.last()['lessons'][0]['code'],'UNCLASSIFIED_MOVEMENT');self.verify()
def test_journal_tables_reject_updates_and_deletes(self):
self.tick()
for table in ('frames','journal_lines','lessons','execution_receipts'):
with self.assertRaises(sqlite3.IntegrityError):self.store.db.execute('DELETE FROM '+table)
self.store.db.rollback()
def test_altered_frame_detected_even_after_trigger_removed(self):
self.tick();db=self.store.db
db.execute('DROP TRIGGER immutable_frames_UPDATE');db.execute("UPDATE frames SET body=replace(body,'BTC/USD','ETH/USD')");db.commit()
with self.assertRaisesRegex(ValueError,'hash chain'):self.verify()
def test_altered_journal_detected(self):
self.tick();db=self.store.db
db.execute('DROP TRIGGER immutable_journal_lines_UPDATE');db.execute("UPDATE journal_lines SET debit='1' WHERE seq=0");db.commit()
with self.assertRaisesRegex(ValueError,'journal/body'):self.verify()
def test_atomic_rollback_if_frame_write_fails(self):
with patch.object(self.e.ledger,'record',side_effect=sqlite3.OperationalError('disk failure')):
with self.assertRaises(sqlite3.OperationalError):self.tick()
self.assertEqual(D(self.store.get('cash')),D(500));self.assertEqual(D(self.store.get('qty')),0)
self.assertTrue(self.store.get('halted'));self.assertEqual(self.store.db.execute('SELECT count(*) FROM frames').fetchone()[0],0)
def test_readonly_view_rejects_writes_and_keeps_database_bytes(self):
self.tick();path=self.root/'backtest.sqlite3';before=path.read_bytes()
db=v.open_readonly(path)
try:
with self.assertRaises(sqlite3.OperationalError):db.execute('DELETE FROM frames')
self.assertEqual(v.verify(db,1)['verified_frames'],1)
finally:db.close()
self.assertEqual(before,path.read_bytes())
def test_viewer_date_window_json_is_complete(self):
self.tick();self.tick(month='1999-05')
script=Path(v.__file__)
result=subprocess.run([sys.executable,str(script),'--project',str(self.root),'--date','1999-04-30'],capture_output=True,text=True)
self.assertEqual(result.returncode,0,result.stderr);data=json.loads(result.stdout)
self.assertEqual(len(data['frames']),1);self.assertEqual(data['metadata']['snapshot']['at'][:7],'1999-04')
self.assertEqual(data['metadata']['integrity']['verified_frames'],1)
def test_viewer_milestone_not_reached_is_error(self):
self.tick();result=subprocess.run([sys.executable,str(Path(v.__file__)),'--project',str(self.root),'--milestone','1000'],capture_output=True,text=True)
self.assertEqual(result.returncode,2);self.assertIn('VIEW_REJECTED',result.stderr)
def test_html_has_no_external_resources_or_scripts(self):
self.tick();out=self.root/'view.html'
result=subprocess.run([sys.executable,str(Path(v.__file__)),'--project',str(self.root),'--format','html','--output',str(out)],capture_output=True,text=True)
self.assertEqual(result.returncode,0,result.stderr);html=out.read_text()
self.assertIn('