#!/usr/bin/env python3 """Century evaluation supervisor. No broker calls; requires an actual host replay adapter.""" import argparse from datetime import datetime, timedelta, timezone from decimal import Decimal, localcontext, InvalidOperation import hashlib from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import importlib import multiprocessing import queue import json from pathlib import Path import signal import sqlite3 import threading import time from zoneinfo import ZoneInfo ROOT=Path(__file__).resolve().parent STOP=threading.Event() def utc():return datetime.now(timezone.utc).isoformat() def decimal(value): if not isinstance(value,str) or len(value)>250:raise ValueError('bounded decimal string required') n=Decimal(value) if not n.is_finite():raise ValueError('finite decimal required') return n def target(attempt,year): if type(attempt) is not int or type(year) is not int or attempt<1 or not 1<=year<=100:raise ValueError('attempt/year range') with localcontext() as ctx: ctx.prec=max(100,attempt+110) return Decimal(500000)*Decimal('1.2')**(attempt-1)*Decimal(3)**(year-1) def deviation(actual,goal): with localcontext() as ctx: ctx.prec=max(100,len(goal.as_tuple().digits)+30) return abs(actual-goal)/goal*100 class Store: def __init__(self,path): self.path=Path(path);self.path.parent.mkdir(parents=True,exist_ok=True) self.lock=threading.RLock() with self.connect() as db: db.executescript(''' PRAGMA journal_mode=WAL; CREATE TABLE IF NOT EXISTS events(id INTEGER PRIMARY KEY,created_at TEXT NOT NULL,kind TEXT NOT NULL,payload TEXT NOT NULL); CREATE TABLE IF NOT EXISTS attempts(id INTEGER PRIMARY KEY,started_at TEXT NOT NULL,finished_at TEXT,status TEXT NOT NULL); CREATE TABLE IF NOT EXISTS metadata(key TEXT PRIMARY KEY,value TEXT NOT NULL); ''') def connect(self):return sqlite3.connect(self.path,timeout=10) def event(self,kind,**payload): row={'created_at':utc(),'kind':kind,**payload} with self.lock,self.connect() as db: db.execute('INSERT INTO events(created_at,kind,payload) VALUES(?,?,?)',(row['created_at'],kind,json.dumps(payload))) print(json.dumps(row),flush=True) return row def lesson(self,code,observed,solution,attempt=None,year=None): return self.event('LESSON',code=code,attempt=attempt,year=year,observed=observed, lesson='Record the discrepancy and preserve evidence before changing the strategy.', proposed_solution=solution,solution_status='PROPOSED_NOT_IMPLEMENTED', strategy_weights_updated=False) def start(self): with self.lock,self.connect() as db: cur=db.execute("INSERT INTO attempts(started_at,status) VALUES(?,'RUNNING')",(utc(),));return cur.lastrowid def finish(self,attempt,status): with self.lock,self.connect() as db:db.execute('UPDATE attempts SET status=?,finished_at=? WHERE id=?',(status,utc(),attempt)) self.event('ATTEMPT_FINISHED',attempt=attempt,status=status) def snapshot(self): with self.lock,self.connect() as db: rows=db.execute('SELECT id,created_at,kind,payload FROM events ORDER BY id DESC LIMIT 200').fetchall() counts=dict(db.execute('SELECT kind,count(*) FROM events GROUP BY kind')) attempts=db.execute('SELECT id,started_at,finished_at,status FROM attempts ORDER BY id DESC LIMIT 25').fetchall() return {'events':[{'id':i,'created_at':t,'kind':k,**json.loads(p)} for i,t,k,p in reversed(rows)], 'counts':counts,'attempts':[dict(zip(('attempt','started_at','finished_at','status'),r)) for r in attempts], 'tolerance_pct':'0.0001','scope':'HOST_ADAPTER_REPORTED_REPLAY; not independently broker reconciled', 'model_training':'Lessons logged; no neural-model training or automatic strategy modification.'} def lessons(self): with self.lock,self.connect() as db:rows=db.execute("SELECT id,created_at,payload FROM events WHERE kind='LESSON' ORDER BY id").fetchall() return [{'id':i,'created_at':t,**json.loads(p)} for i,t,p in rows] class Blocked(ValueError):pass def load_adapter(name): if not name:raise Blocked('REPLAY_ADAPTER_NOT_CONFIGURED') adapter=importlib.import_module(name) if not callable(getattr(adapter,'run_year',None)):raise Blocked('REPLAY_ADAPTER_INTERFACE_MISSING') return adapter def validate_report(report,calendar_year): if not isinstance(report,dict):raise Blocked('INVALID_YEAR_REPORT') if report.get('data_origin')!='HISTORICAL_REPLAY':raise Blocked('HISTORICAL_REPLAY_EVIDENCE_REQUIRED') if report.get('calendar_year')!=calendar_year:raise Blocked('YEAR_MISMATCH') if report.get('starting_capital_usd')!='500':raise Blocked('ANNUAL_STARTING_CAPITAL_MISMATCH') income=decimal(report.get('net_income_usd')) if report.get('income_definition')!='REALIZED_NET_AFTER_COSTS':raise Blocked('INCOME_DEFINITION_MISMATCH') controls=report.get('controls',{}) for key in ('intraday_3pct_breaker','capital_floor_ratchet','no_live_orders','point_in_time_inputs'): if controls.get(key) is not True:raise Blocked('CONTROL_NOT_CONFIRMED:'+key) for key in ('strategy_sha256','dataset_sha256','ledger_sha256'): h=report.get(key) if not isinstance(h,str) or len(h)!=64 or any(c not in '0123456789abcdef' for c in h):raise Blocked('EVIDENCE_HASH_REQUIRED:'+key) sources=report.get('sources') if not isinstance(sources,list) or not sources:raise Blocked('SOURCE_PROVENANCE_REQUIRED') for source in sources: if not isinstance(source,dict) or not isinstance(source.get('url'),str) or not source['url'].startswith('https://'): raise Blocked('SOURCE_PROVENANCE_INVALID') if source.get('point_in_time_validated') is not True:raise Blocked('POINT_IN_TIME_NOT_VALIDATED') return income def replay_worker(module_name,context,result_queue): try: adapter=importlib.import_module(module_name) result_queue.put(('ok',adapter.run_year(context))) except Exception as exc:result_queue.put(('error',type(exc).__name__)) def execute_year(adapter,context,deadline): """Isolate the replay worker so stop/deadline can terminate a hung year.""" mp=multiprocessing.get_context('spawn');results=mp.Queue() worker=mp.Process(target=replay_worker,args=(adapter.__name__,context,results)) worker.start() try: while True: if STOP.is_set() or (deadline and datetime.now(timezone.utc)>=deadline): raise Blocked('RUN_DEADLINE_OR_STOP') try:status,value=results.get(timeout=.25) except queue.Empty: if not worker.is_alive():raise Blocked('REPLAY_WORKER_EXITED_WITHOUT_REPORT') continue if status!='ok':raise Blocked('REPLAY_ADAPTER_ERROR:'+value) return value finally: if worker.is_alive():worker.terminate() worker.join(timeout=2) if worker.is_alive():worker.kill();worker.join(timeout=2) results.close() def run_attempt(store,adapter,start_year,deadline=None): attempt=store.start();passed=True store.event('ATTEMPT_STARTED',attempt=attempt,first_calendar_year=start_year,years=100, starting_capital_each_year_usd='500',year1_target_usd=str(target(attempt,1)),target_deviation_tolerance_pct='0.0001') try: for year in range(1,101): if STOP.is_set() or (deadline and datetime.now(timezone.utc)>=deadline): store.finish(attempt,'INTERRUPTED');return False goal=target(attempt,year) context={'attempt':attempt,'year':year,'calendar_year':start_year+year-1, 'starting_capital_usd':'500','income_target_usd':str(goal),'tolerance_pct':'0.0001', 'protective_controls':{'intraday_drawdown_pct':'3','capital_floor_ratchet_required':True}, 'deadline_utc':deadline.isoformat() if deadline else None, 'lesson_history':store.lessons(), 'mode':'HISTORICAL_REPLAY_ONLY','allow_live_orders':False} store.event('YEAR_STARTED',attempt=attempt,year=year,calendar_year=context['calendar_year'],income_target_usd=str(goal)) report=execute_year(adapter,context,deadline) if deadline and datetime.now(timezone.utc)>=deadline: store.finish(attempt,'INTERRUPTED');return False actual=validate_report(report,context['calendar_year']);error=deviation(actual,goal) ok=error<=Decimal('0.0001');passed=passed and ok store.event('YEAR_RESULT',attempt=attempt,year=year,calendar_year=context['calendar_year'], net_income_usd=str(actual),income_target_usd=str(goal),target_deviation_pct=str(error), accepted=ok,measurement_origin='HOST_ADAPTER_REPORTED', evidence={k:report[k] for k in ('strategy_sha256','dataset_sha256','ledger_sha256','sources','controls')}) if not ok:store.lesson('TARGET_DEVIATION_EXCEEDED', {'actual_income_usd':str(actual),'target_income_usd':str(goal),'deviation_pct':str(error),'threshold_pct':'0.0001'}, 'Reconcile the replay ledger, fees, slippage and opportunity/exposure limits; test a specific strategy change on separate held-out periods before accepting it.',attempt,year) store.finish(attempt,'PASSED' if passed else 'FAILED');return passed except Exception as exc: # Do not echo private adapter exception messages, request headers or credentials. code=str(exc) if isinstance(exc,Blocked) else type(exc).__name__ store.event('RUN_BLOCKED',attempt=attempt,code=code) store.lesson('REPLAY_BLOCKED',{'code':code},'Connect the actual historical replay adapter and timestamped market/news datasets; verify source and control evidence, then rerun.',attempt) store.finish(attempt,'BLOCKED');return False def sunday_deadline(): now=datetime.now(ZoneInfo('America/New_York')) day=now.date()+timedelta(days=(6-now.weekday())%7) return datetime.combine(day,datetime.max.time(),now.tzinfo).astimezone(timezone.utc) def make_server(store,port): class Handler(BaseHTTPRequestHandler): def log_message(self,*args):pass def do_GET(self): if self.path=='/':body=(ROOT/'monitor.html').read_bytes();kind='text/html; charset=utf-8' elif self.path=='/api/status':body=json.dumps(store.snapshot()).encode();kind='application/json' elif self.path=='/api/lessons':body=json.dumps(store.lessons()).encode();kind='application/json' else:self.send_error(404);return self.send_response(200);self.send_header('Content-Type',kind);self.send_header('Content-Length',str(len(body))) self.send_header('Cache-Control','no-store');self.send_header('X-Content-Type-Options','nosniff');self.end_headers();self.wfile.write(body) return ThreadingHTTPServer(('127.0.0.1',port),Handler) def main(): p=argparse.ArgumentParser(description=__doc__) p.add_argument('--adapter',help='Local module providing run_year(context); never a broker order adapter') p.add_argument('--start-year',type=int,default=1926,help='100 complete years, default 1926..2025') p.add_argument('--db',type=Path,default=ROOT/'runs'/'gauntlet.db') p.add_argument('--continuous',action='store_true') p.add_argument('--until',help='Timezone-aware ISO datetime; default Sunday night New York') p.add_argument('--interval',type=float,default=60) p.add_argument('--monitor-only',action='store_true');p.add_argument('--serve',action='store_true');p.add_argument('--port',type=int,default=8890) args=p.parse_args() if args.interval<1 or not 1<=args.start_year<=9900:p.error('invalid interval or year') until=datetime.fromisoformat(args.until.replace('Z','+00:00')) if args.until else sunday_deadline() if until.tzinfo is None:p.error('--until requires timezone') store=Store(args.db) signal.signal(signal.SIGTERM,lambda *_:STOP.set());signal.signal(signal.SIGINT,lambda *_:STOP.set()) server=None if args.serve or args.monitor_only: server=make_server(store,args.port);threading.Thread(target=server.serve_forever,daemon=True).start() store.event('MONITOR_STARTED',url=f'http://127.0.0.1:{server.server_address[1]}',remote_access='Use existing authenticated gateway; do not expose this unauthenticated local monitor directly.') try: if args.monitor_only: while not STOP.is_set() and datetime.now(timezone.utc)