#!/usr/bin/env python3
"""FinBot Century Replay: one-file, native historical/synthetic simulation.
Run in the existing project: python3 FinBot_Century_Replay.py
Verify fixtures only: python3 FinBot_Century_Replay.py --self-test
Default window: 1926-01-01 through 2025-12-31, exactly 100 calendar years.
Optional: --continuous repeats complete attempts until interrupted.
This file contains the complete execution core, adapter and supervisor as
plain Python source sections. No installer, downloaded code or binary payload.
It loads your existing host strategy and normalized data without editing them.
No substitute strategy is selected for real runs. --self-test alone uses a fixture.
Synthetic monthly curves are modeled scenarios, not actual historical returns.
Missing inputs are explicitly logged; a flat missing-input month adds no evidence.
Raw retrospective GDELT ZIP/CSV and ALFRED CSV require normalized JSONL input.
The existing bundle's raw 1979 archive is not represented as century news coverage.
Monthly risk checks cannot establish intraday drawdown protection.
Existing halts persist through handoff; there is no automatic strategy retraining.
This is an explicit native simulation entrypoint, not the Docker-protected
deployment. It does not replace or weaken the existing gateway, Docker runner,
dashboard sessions or live broker engine. Only use the trusted local strategy.
It starts no web server and contains no live broker order submission function.
The private host strategy's behavior is outside the fixture verification scope.
"""
import hashlib
import json
import math
import os
from pathlib import Path
import sys
from datetime import datetime, timezone
COMPONENTS = {}
# BEGIN COMPLETE SOURCE: _finbot_privacy
COMPONENTS['_finbot_privacy'] = """\"\"\"Redact credential fields and URL authentication material before telemetry.\"\"\"
import re
from urllib.parse import urlsplit,urlunsplit
SECRET=re.compile(r'password|secret|token|authorization|cookie|api.?key|credential|private.?key',re.I)
def clean(value):
if isinstance(value,dict):return {str(k):('[REDACTED]' if SECRET.search(str(k)) else clean(v)) for k,v in value.items()}
if isinstance(value,(list,tuple)):return [clean(v) for v in value]
if isinstance(value,str) and value.startswith(('https://','http://')):
try:
u=urlsplit(value);host=u.hostname or '';port=u.port
if ':' in host:host='['+host+']'
return urlunsplit((u.scheme,host+(':'+str(port) if port else ''),u.path,'',''))
except ValueError:return '[INVALID URL]'
return value
# Mask recognized inline credentials and exact secret values present in this
# process environment. Unknown secrets in arbitrary prose cannot be guaranteed.
_original_clean=clean
_INLINE=re.compile(r'(?i)(bearer\\s+)[A-Za-z0-9._~+/=-]+|((?:password|secret|api[_ -]?key|access[_ -]?token)\\s*[:=]\\s*)[^\\s,;]+')
def clean(value):
import os
if isinstance(value,str):
value=_INLINE.sub(lambda m:(m.group(1) or m.group(2) or '')+'[REDACTED]',value)
for key,secret in os.environ.items():
if SECRET.search(key) and len(secret)>=8:value=value.replace(secret,'[REDACTED]')
return _original_clean(value)
"""
# END COMPLETE SOURCE: _finbot_privacy
# BEGIN COMPLETE SOURCE: _finbot_core
COMPONENTS['_finbot_core'] = """\"\"\"Decimal-only, long-only local paper simulator. Never submits broker orders.\"\"\"
from __future__ import annotations
import calendar
import hashlib
import json
import os
import sqlite3
from datetime import date, datetime, timedelta, timezone
from decimal import Decimal, InvalidOperation, ROUND_FLOOR
from pathlib import Path
UTC = timezone.utc
CHANNELS = ('geopolitical', 'macroeconomic', 'behavioral', 'accounting', 'tax', 'quantitative')
class Rejected(ValueError):
pass
def number(value, positive=False):
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError):
raise Rejected('Invalid decimal') from None
if not result.is_finite() or result < 0 or (positive and result <= 0):
raise Rejected('Expected finite nonnegative number')
return result
def instant(value):
try:
result = datetime.fromisoformat(str(value).replace('Z', '+00:00'))
except ValueError:
raise Rejected('Invalid timestamp') from None
if result.tzinfo is None:
raise Rejected('Timestamp must include timezone')
return result.astimezone(UTC)
def timestamp():
return datetime.now(UTC).isoformat()
def holding_class(acquired, sold):
\"\"\"Calendar anniversary, not a fixed 365-day approximation; plain capital assets.\"\"\"
a, s = date.fromisoformat(acquired), date.fromisoformat(sold)
if s < a:
raise Rejected('Disposition predates acquisition')
anniversary = date(a.year + 1, a.month, min(a.day, calendar.monthrange(a.year + 1, a.month)[1]))
return 'LONG_TERM' if s > anniversary else 'SHORT_TERM'
def volume_breaker(history, current):
if len(history) != 20:
raise Rejected('Exactly 20 preceding closed bars required')
h = [number(x) for x in history]
v = number(current)
mean = sum(h) / 20
variance = sum((x - mean) ** 2 for x in h) / 20
if variance == 0:
return {'z_score': None, 'action': 'HALT_UNCALIBRATED_BASELINE'}
z = (v - mean) / variance.sqrt()
return {'z_score': str(z), 'action': 'HALT_REVIEW' if z >= 5 else 'FREEZE_ENTRIES' if z >= 3 else 'NORMAL'}
SQLITE = '''
CREATE TABLE IF NOT EXISTS accounts(
id INTEGER PRIMARY KEY CHECK(id=1), cash TEXT NOT NULL, initial_equity TEXT NOT NULL,
session TEXT NOT NULL, session_equity TEXT NOT NULL, peak_equity TEXT NOT NULL,
halted INTEGER NOT NULL CHECK(halted IN(0,1)), started_at TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS positions(
id INTEGER PRIMARY KEY, symbol TEXT NOT NULL UNIQUE, sector TEXT NOT NULL,
quantity TEXT NOT NULL, entry TEXT NOT NULL, stop TEXT NOT NULL, mark TEXT NOT NULL,
fee TEXT NOT NULL, acquired_at TEXT NOT NULL, quote_at TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS orders(
id INTEGER PRIMARY KEY, request_id TEXT NOT NULL UNIQUE, payload_hash TEXT NOT NULL,
symbol TEXT NOT NULL, side TEXT NOT NULL CHECK(side IN('BUY','SELL')),
quantity TEXT NOT NULL, price TEXT NOT NULL, fee TEXT NOT NULL, notional TEXT NOT NULL,
execution_tag TEXT NOT NULL DEFAULT 'LOCAL_SIMULATION', executed_at TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS reasons(
order_id INTEGER PRIMARY KEY REFERENCES orders(id), evaluation_json TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS position_policies(
position_id INTEGER PRIMARY KEY REFERENCES positions(id) ON DELETE CASCADE,
entry_order_id INTEGER NOT NULL UNIQUE REFERENCES orders(id),
execution_tag TEXT NOT NULL, trailing_pct TEXT NOT NULL);
CREATE INDEX IF NOT EXISTS idx_position_tags ON position_policies(execution_tag);
CREATE TABLE IF NOT EXISTS dispositions(
id INTEGER PRIMARY KEY, sell_order_id INTEGER NOT NULL UNIQUE REFERENCES orders(id),
symbol TEXT NOT NULL, quantity TEXT NOT NULL, acquired_at TEXT NOT NULL,
sold_at TEXT NOT NULL, proceeds TEXT NOT NULL, basis TEXT NOT NULL,
economic_pnl TEXT NOT NULL, holding_class TEXT NOT NULL, review_status TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS snapshots(
id INTEGER PRIMARY KEY, captured_at TEXT NOT NULL, equity TEXT NOT NULL,
cash TEXT NOT NULL, open_count INTEGER NOT NULL, drawdown TEXT NOT NULL);
CREATE TABLE IF NOT EXISTS heartbeats(
worker TEXT PRIMARY KEY, observed_at TEXT NOT NULL, status TEXT NOT NULL);
'''
class Engine:
def __init__(self, path=None):
if os.environ.get('TRADING_MODE', 'LOCAL_SIMULATION') != 'LOCAL_SIMULATION':
raise Rejected('This release only permits LOCAL_SIMULATION; no broker execution')
if os.environ.get('ALPACA_BASE_URL', 'https://paper-api.alpaca.markets') != 'https://paper-api.alpaca.markets':
raise Rejected('Live or alternate broker URL prohibited')
path = path or os.environ.get('QUANT_DB_PATH', str(Path.home() / '.local/share/foresixa-quant/ledger.sqlite3'))
if path != ':memory:':
Path(path).parent.mkdir(parents=True, exist_ok=True, mode=0o700)
self.db = sqlite3.connect(path, timeout=10, isolation_level=None)
self.db.row_factory = sqlite3.Row
fixture_marker = self.db.execute("SELECT 1 FROM sqlite_master WHERE type='table' AND name='sandbox_core_metadata'").fetchone()
if fixture_marker and not getattr(self, '_allows_synthetic_fixture', False):
self.db.close()
raise Rejected('Synthetic core fixture database cannot be used as the strategy ledger')
self.db.execute('PRAGMA foreign_keys=ON')
self.db.execute('PRAGMA journal_mode=WAL')
self.db.executescript(SQLITE)
if path != ':memory:':
os.chmod(path, 0o600)
def initialize(self, virtual_cash='25000'):
amount = number(virtual_cash, True)
with self.transaction():
if self.db.execute('SELECT 1 FROM accounts').fetchone():
raise Rejected('Existing account cannot be reset through initialize')
now = timestamp()
self.db.execute('INSERT INTO accounts VALUES(1,?,?,?,?,?,0,?)',
(str(amount), str(amount), now[:10], str(amount), str(amount), now))
return self.status()
def transaction(self):
from contextlib import contextmanager
@contextmanager
def atomic():
self.db.execute('BEGIN IMMEDIATE')
try:
yield
except BaseException:
self.db.rollback()
raise
else:
self.db.commit()
return atomic()
def account(self):
a = self.db.execute('SELECT * FROM accounts WHERE id=1').fetchone()
if a is None:
raise Rejected('Initialize explicit virtual account first')
return a
def positions(self):
return self.db.execute('SELECT * FROM positions').fetchall()
def equity(self):
return number(self.account()['cash']) + sum(number(p['quantity']) * number(p['mark']) for p in self.positions())
def fresh(self, at, max_seconds=60):
age = (datetime.now(UTC) - instant(at)).total_seconds()
if age < -5 or age > max_seconds:
raise Rejected('Stale or future-dated data')
def guards(self):
a = self.account()
for p in self.positions():
self.fresh(p['quote_at'])
eq = self.equity()
today = timestamp()[:10]
if a['session'] != today and not a['halted']:
self.db.execute('UPDATE accounts SET session=?,session_equity=? WHERE id=1', (today, str(eq)))
a = self.account()
drawdown = max(Decimal(0), (number(a['session_equity'], True) - eq) / number(a['session_equity'], True))
peak = max(eq, number(a['peak_equity'], True))
trailing = (peak - eq) / peak
if drawdown >= Decimal('.03') or trailing >= Decimal('.10'):
self.db.execute('UPDATE accounts SET halted=1 WHERE id=1')
self.db.execute('UPDATE accounts SET peak_equity=? WHERE id=1', (str(peak),))
return eq, drawdown, trailing
def research(self, r):
if not isinstance(r, dict) or r.get('decision') != 'RESEARCH_PASSED':
raise Rejected('Evidence-backed RESEARCH_PASSED evaluation required')
self.fresh(r.get('analyzed_at'), 3600)
for channel in CHANNELS:
v = r.get(channel)
if not isinstance(v, dict) or v.get('status') != 'PASS' or not isinstance(v.get('summary'), str) or not v['summary'].strip():
raise Rejected('Missing research gate: ' + channel)
if not isinstance(v.get('sources'), list) or not v['sources']:
raise Rejected('Missing sources: ' + channel)
for s in v['sources']:
if not isinstance(s, dict) or not str(s.get('url', '')).startswith('https://') or not s.get('observed_at'):
raise Rejected('Source URL and observed_at required')
instant(s['observed_at'])
# Structural completeness is not independent verification of any assertion.
return r
def evaluate(self, symbol, entry, stop, quote_at, reasoning, sector='DIGITAL_ASSETS'):
if symbol not in ('BTC/USD', 'ETH/USD'):
raise Rejected('Symbol outside explicit BTC/USD,ETH/USD universe')
if sector != 'DIGITAL_ASSETS':
raise Rejected('Crypto sector cannot be relabeled to bypass concentration')
price, stop = number(entry, True), number(stop, True)
if stop >= price:
raise Rejected('Long-only stop must be below entry')
self.fresh(quote_at)
self.research(reasoning)
eq, dd, trailing = self.guards()
if self.account()['halted']:
raise Rejected('Account halted by drawdown or recovery gate')
if any(p['symbol'] == symbol for p in self.positions()):
raise Rejected('One open lot per symbol in this release')
cash = number(self.account()['cash'])
deployed = sum(number(p['quantity']) * number(p['mark']) for p in self.positions())
fee_rate = Decimal('.0025')
slippage = Decimal('.001')
# Include modeled entry/exit fees and slippage in planned loss budget.
risk_per_unit = price - stop + price * (2 * fee_rate + 2 * slippage)
quantity = min(eq * Decimal('.01') / risk_per_unit,
cash / (price * (1 + fee_rate)),
max(Decimal(0), eq - deployed) / price,
max(Decimal(0), eq * Decimal('.25') - deployed) / (price * (1 + Decimal('.25') * fee_rate)))
quantity = quantity.quantize(Decimal('.00000001'), rounding=ROUND_FLOOR)
if quantity <= 0:
raise Rejected('No capacity under cash, exposure and concentration caps')
return {'status': 'LOCAL_SIMULATION_ELIGIBLE', 'symbol': symbol, 'quantity': str(quantity),
'entry': str(price), 'stop': str(stop), 'notional': str(quantity * price),
'entry_fee': str(quantity * price * fee_rate), 'planned_risk': str(quantity * risk_per_unit),
'risk_cap': str(eq * Decimal('.01')), 'daily_drawdown': str(dd),
'trailing_drawdown': str(trailing), 'loss_guaranteed': False}
def simulate_entry(self, request_id, symbol, entry, stop, quote_at, reasoning, requested_qty=None, trailing_pct=None, execution_tag=None, risk_limits=None):
if not isinstance(request_id, str) or not 1 <= len(request_id) <= 100:
raise Rejected('Request ID required, maximum 100 characters')
payload = [symbol, entry, stop, quote_at, reasoning]
if requested_qty is not None:
payload.append(str(number(requested_qty, True)))
if trailing_pct is not None:
trailing_pct = number(trailing_pct, True)
if trailing_pct >= 1 or number(stop, True) != number(entry, True)*(1-trailing_pct):
raise Rejected('Trailing distance must match the initial protective stop')
if not isinstance(execution_tag, str) or not 1 <= len(execution_tag) <= 50:
raise Rejected('Execution tag required for trailing position')
payload.append({'trailing_pct': str(trailing_pct), 'execution_tag': execution_tag})
elif execution_tag is not None:
raise Rejected('Tagged entry requires a trailing policy')
if risk_limits is not None:
if not isinstance(risk_limits, dict) or set(risk_limits) != {'per_trade_rate','snapshot_cap'}:
raise Rejected('Invalid additional risk limits')
rate, budget = number(risk_limits['per_trade_rate'], True), number(risk_limits['snapshot_cap'], True)
if rate > Decimal('.01'):
raise Rejected('Additional risk rate cannot exceed 1%')
payload.append({'per_trade_rate':str(rate),'snapshot_cap':str(budget)})
digest = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
with self.transaction():
old = self.db.execute('SELECT * FROM orders WHERE request_id=?', (request_id,)).fetchone()
if old:
if old['payload_hash'] != digest:
raise Rejected('Idempotency key reused with different request')
return {'status': 'ALREADY_RECORDED', 'order_id': old['id']}
e = self.evaluate(symbol, entry, stop, quote_at, reasoning)
if risk_limits is not None:
previous_q = number(e['quantity'], True)
cap = min(number(e['risk_cap']), number(e['risk_cap'])*rate/Decimal('.01'), budget)
q = min(previous_q, previous_q*cap/number(e['planned_risk'], True)).quantize(Decimal('.00000001'), rounding=ROUND_FLOOR)
if q <= 0:
raise Rejected('Additional risk limits allow no quantity')
e.update(quantity=str(q),notional=str(q*number(e['entry'])),
entry_fee=str(q*number(e['entry'])*Decimal('.0025')),
planned_risk=str(q*number(e['planned_risk'])/previous_q),risk_cap=str(cap))
if requested_qty is not None:
previous_q = number(e['quantity'], True)
q = min(previous_q, number(requested_qty, True)).quantize(Decimal('.00000001'), rounding=ROUND_FLOOR)
if q <= 0:
raise Rejected('Requested quantity below simulation increment')
e.update(quantity=str(q), notional=str(q*number(e['entry'])),
entry_fee=str(q*number(e['entry'])*Decimal('.0025')),
planned_risk=str(q*number(e['planned_risk'])/previous_q))
now = timestamp()
q, p, fee = number(e['quantity']), number(e['entry']), number(e['entry_fee'])
cur = self.db.execute('INSERT INTO orders(request_id,payload_hash,symbol,side,quantity,price,fee,notional,executed_at) VALUES(?,?,?,\\'BUY\\',?,?,?,?,?)',
(request_id, digest, symbol, str(q), str(p), str(fee), str(q*p), now))
oid = cur.lastrowid
self.db.execute('INSERT INTO reasons VALUES(?,?)', (oid, json.dumps(reasoning, sort_keys=True)))
position_cursor = self.db.execute('INSERT INTO positions(symbol,sector,quantity,entry,stop,mark,fee,acquired_at,quote_at) VALUES(?,\\'DIGITAL_ASSETS\\',?,?,?,?,?,?,?)',
(symbol, str(q), str(p), e['stop'], str(p), str(fee), now, quote_at))
if trailing_pct is not None:
self.db.execute('INSERT INTO position_policies VALUES(?,?,?,?)',
(position_cursor.lastrowid, oid, execution_tag, str(trailing_pct)))
self.db.execute('UPDATE orders SET execution_tag=? WHERE id=?', (execution_tag, oid))
self.db.execute('UPDATE accounts SET cash=? WHERE id=1', (str(number(self.account()['cash']) - q*p - fee),))
self.snapshot()
return {**e, 'order_id': oid, 'execution_mode': 'LOCAL_SIMULATION', 'broker_submitted': False}
def close_position(self, symbol, price, at, cause='LOCAL_EXIT'):
p = self.db.execute('SELECT * FROM positions WHERE symbol=?', (symbol,)).fetchone()
if p is None:
raise Rejected('No position; short selling prohibited')
price, q = number(price, True), number(p['quantity'], True)
self.fresh(at)
fee = q * price * Decimal('.0025')
proceeds, basis = q*price-fee, q*number(p['entry'])+number(p['fee'])
now = timestamp()
key = f'exit-{p["id"]}-{p["acquired_at"]}'
cur = self.db.execute('INSERT INTO orders(request_id,payload_hash,symbol,side,quantity,price,fee,notional,executed_at) VALUES(?,?,?,\\'SELL\\',?,?,?,?,?)',
(key, cause, symbol, str(q), str(price), str(fee), str(q*price), now))
policy = self.db.execute('SELECT * FROM position_policies WHERE position_id=?', (p['id'],)).fetchone()
if policy:
self.db.execute('UPDATE orders SET execution_tag=? WHERE id=?', (policy['execution_tag'], cur.lastrowid))
self.db.execute('INSERT INTO dispositions(sell_order_id,symbol,quantity,acquired_at,sold_at,proceeds,basis,economic_pnl,holding_class,review_status) VALUES(?,?,?,?,?,?,?,?,?,\\'SIMULATION_NOT_TAX_FILING\\')',
(cur.lastrowid, symbol, str(q), p['acquired_at'], now, str(proceeds), str(basis), str(proceeds-basis), holding_class(p['acquired_at'][:10], now[:10])))
self.db.execute('DELETE FROM positions WHERE symbol=?', (symbol,))
self.db.execute('UPDATE accounts SET cash=? WHERE id=1', (str(number(self.account()['cash']) + proceeds),))
return {'symbol': symbol, 'economic_pnl': str(proceeds-basis), 'cause': cause}
def mark(self, quotes):
if not isinstance(quotes, dict):
raise Rejected('Quotes must be a symbol-indexed object')
with self.transaction():
for p in self.positions():
if p['symbol'] not in quotes:
raise Rejected('Complete mark set required for all open positions')
for symbol, quote in quotes.items():
if symbol not in ('BTC/USD', 'ETH/USD'):
raise Rejected('Unauthorized symbol')
self.fresh(quote['at'])
price = number(quote['price'], True)
self.db.execute('UPDATE positions SET mark=?,quote_at=? WHERE symbol=?', (str(price), quote['at'], symbol))
eq, dd, trailing = self.guards()
halted = bool(self.account()['halted'])
exits, stop_updates = [], []
for p in self.positions():
if halted or number(p['mark']) <= number(p['stop']):
# Simulated exit at observed price, including gap through stop and fee.
exits.append(self.close_position(p['symbol'], p['mark'], p['quote_at'], 'DRAWDOWN_HALT' if halted else 'STOP_TRIGGER'))
else:
policy = self.db.execute('SELECT * FROM position_policies WHERE position_id=?', (p['id'],)).fetchone()
if policy:
next_stop = number(p['mark'])*(1-number(policy['trailing_pct'], True))
if next_stop > number(p['stop']):
self.db.execute('UPDATE positions SET stop=? WHERE id=?', (str(next_stop), p['id']))
stop_updates.append({'symbol': p['symbol'], 'previous_stop': p['stop'], 'stop': str(next_stop)})
# Closing fees can push realized equity across a threshold.
_, dd, trailing = self.guards()
halted = bool(self.account()['halted'])
if halted and self.positions():
for p in self.positions():
exits.append(self.close_position(p['symbol'], p['mark'], p['quote_at'], 'DRAWDOWN_HALT'))
self.snapshot()
return {'status': 'HALTED' if halted else 'MARKED', 'exits': exits, 'stop_updates': stop_updates, 'daily_drawdown': str(dd), 'trailing_drawdown': str(trailing)}
def recover(self):
with self.transaction():
a = self.account()
if a['session'] == timestamp()[:10] or self.positions():
raise Rejected('Recovery requires a later UTC session and no open positions')
eq = self.equity()
if (number(a['peak_equity']) - eq) / number(a['peak_equity'], True) >= Decimal('.10'):
raise Rejected('10% trailing drawdown requires separate strategy review; cannot reset peak')
self.db.execute('UPDATE accounts SET halted=0,session=?,session_equity=? WHERE id=1', (timestamp()[:10], str(eq)))
return self.status()
def snapshot(self):
a = self.account()
eq = self.equity()
peak = max(number(a['peak_equity'], True), eq)
self.db.execute('UPDATE accounts SET peak_equity=? WHERE id=1', (str(peak),))
self.db.execute('INSERT INTO snapshots(captured_at,equity,cash,open_count,drawdown) VALUES(?,?,?,?,?)',
(timestamp(), str(eq), a['cash'], len(self.positions()), str((peak-eq)/peak)))
def heartbeat(self, worker='crypto-collector', status='IDLE'):
if worker != 'crypto-collector' or status not in ('IDLE','FEED_OK','FEED_ERROR'):
raise Rejected('Invalid worker heartbeat')
self.db.execute('INSERT INTO heartbeats VALUES(?,?,?) ON CONFLICT(worker) DO UPDATE SET observed_at=excluded.observed_at,status=excluded.status', (worker, timestamp(), status))
def status(self):
a = self.db.execute('SELECT * FROM accounts').fetchone()
h = self.db.execute('SELECT * FROM heartbeats').fetchall()
return {'mode': 'LOCAL_SIMULATION', 'live_execution_enabled': False, 'broker_orders_enabled': False,
'futures': 'DISABLED_UNVERIFIED_ADAPTER', 'account_initialized': a is not None,
'equity': str(self.equity()) if a else None, 'cash': a['cash'] if a else None,
'halted': bool(a['halted']) if a else None, 'positions': [dict(p) for p in self.positions()],
'workers': [{**dict(r), 'fresh': (datetime.now(UTC)-instant(r['observed_at'])).total_seconds() <= 300} for r in h],
'trial_started': False, 'trial_reason': 'No approved autonomous strategy or broker integration in this release'}
def report(self, days=7):
if not isinstance(days, int) or isinstance(days, bool) or not 1 <= days <= 3660:
raise Rejected('Days must be 1..3660')
start = (datetime.now(UTC)-timedelta(days=days)).isoformat()
rows = self.db.execute('SELECT * FROM dispositions WHERE sold_at>=?', (start,)).fetchall()
pnl = [Decimal(r['economic_pnl']) for r in rows]
wins = sum((x for x in pnl if x>0), Decimal(0))
losses = -sum((x for x in pnl if x<0), Decimal(0))
pf = wins/losses if losses else None
return {'scope': 'LOCAL_SIMULATION_CLOSED_LOTS', 'closed_lots': len(rows), 'net_pnl': str(sum(pnl,Decimal(0))),
'gross_profits': str(wins), 'gross_losses': str(losses), 'profit_factor': str(pf) if pf is not None else None,
'profit_factor_status': 'DEFINED' if losses else 'UNDEFINED_NO_LOSSES', 'tax_adjustments_in_performance': False}
def audit(self):
metrics = self.report(3660)
return {'status': 'REVIEW_REQUIRED_LIVE_DISABLED', 'metrics': metrics,
'trial_days': 0, 'requirements': {'minimum_continuous_verified_days': 90, 'minimum_closed_lots_per_worker': 100, 'profit_factor_target': '1.5'},
'unverified': ['autonomous strategy', 'broker fill reconciliation', 'continuous 90-day observations', 'out-of-sample robustness', 'operational recovery', 'legal/tax review'],
'real_cash_activation': 'UNIMPLEMENTED'}
def tax_export(self, year):
if not isinstance(year, int) or isinstance(year, bool) or not 1900 <= year <= 9998:
raise Rejected('Tax year outside supported range')
rows = self.db.execute('SELECT * FROM dispositions WHERE sold_at>=? AND sold_at250: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<=1000: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,mode=0o700)
if self.path.is_symlink():raise ValueError('Database symlink rejected')
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):
db=sqlite3.connect(self.path,timeout=10)
os.chmod(self.path,0o600)
return db
def event(self,kind,**payload):
from _finbot_privacy import clean
payload=clean(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,**clean(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 activity(self,after=0,before=None,q='',kind='',limit=250):
if after<0 or (before is not None and before<1) or not 1<=limit<=1000:raise ValueError('Invalid cursor/limit')
clauses=['id>?'];params=[after]
if before is not None:clauses.append('id0");params.append(q.lower()[:300])
if kind:clauses.append('kind=?');params.append(kind[:100])
order='DESC' if before is not None else 'ASC'
with self.lock,self.connect() as db:
rows=db.execute('SELECT id,created_at,kind,payload FROM events WHERE '+' AND '.join(clauses)+' ORDER BY id '+order+' LIMIT ?',params+[limit+1]).fetchall()
more=len(rows)>limit;rows=rows[:limit]
if before is not None:rows.reverse()
return {'events':[{'id':i,'created_at':t,'kind':k,**clean(json.loads(p))} for i,t,k,p in rows],
'has_more':more,'next_after':rows[-1][0] if rows else after,'next_before':rows[0][0] if rows else before}
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,**clean(json.loads(p))} for i,t,p in rows]
class Blocked(ValueError):pass
def checkpoint_number(value):
if not isinstance(value, str) or len(value) > 4096:
raise Blocked('CHECKPOINT_DECIMAL_STRING_REQUIRED')
try:
result = Decimal(value)
except InvalidOperation:
raise Blocked('CHECKPOINT_INVALID_DECIMAL') from None
if not result.is_finite() or abs(result.adjusted()) > 3000:
raise Blocked('CHECKPOINT_NONFINITE_OR_OVERSIZED')
return result
def checkpoint_time(value):
if not isinstance(value, str):
raise Blocked('CHECKPOINT_TIMESTAMP_REQUIRED')
try:
at = datetime.fromisoformat(value.replace('Z', '+00:00'))
except ValueError:
raise Blocked('CHECKPOINT_TIMESTAMP_INVALID') from None
if at.tzinfo is None:
raise Blocked('CHECKPOINT_TIMEZONE_REQUIRED')
return at.astimezone(timezone.utc)
def validate_checkpoint(cp):
if not isinstance(cp, dict) or cp.get('version') != 1:
raise Blocked('CHECKPOINT_VERSION_REQUIRED')
required = ('as_of', 'cash_usd', 'equity_usd', 'positions', 'locked_floor_usd',
'peak_equity_usd', 'initial_capital_usd', 'cumulative_net_income_usd',
'halted', 'session', 'session_equity_usd')
if any(key not in cp for key in required):
raise Blocked('CHECKPOINT_FIELDS_MISSING')
if cp['positions'] != [] or type(cp['halted']) is not bool:
raise Blocked('CLOSED_CHECKPOINT_REQUIRED')
at = checkpoint_time(cp['as_of'])
cash, equity, floor, peak, initial, cumulative, session_equity = (
checkpoint_number(cp[key]) for key in ('cash_usd', 'equity_usd', 'locked_floor_usd',
'peak_equity_usd', 'initial_capital_usd', 'cumulative_net_income_usd', 'session_equity_usd'))
with localcontext() as precision:
precision.prec = 4096
if cash < 0 or cash != equity or initial != 500 or equity - initial != cumulative:
raise Blocked('CHECKPOINT_BALANCE_INVALID')
if not 0 <= floor <= peak or peak < equity or peak <= 0 or session_equity <= 0:
raise Blocked('CHECKPOINT_RISK_STATE_INVALID')
if cp['session'] != at.date().isoformat():
raise Blocked('CHECKPOINT_SESSION_INVALID')
return dict(cp)
def validate_annual_evidence(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 validate_compounded_result(report, calendar_year, incoming):
if not isinstance(report, dict):
raise Blocked('INVALID_YEAR_REPORT')
with localcontext() as precision:
precision.prec = 4096
outgoing = validate_checkpoint(report.get('account_checkpoint'))
start = checkpoint_number(report.get('starting_capital_usd'))
expected = checkpoint_number(incoming['cash_usd']) if incoming else Decimal(500)
income = checkpoint_number(report.get('net_income_usd'))
cash = checkpoint_number(report.get('ending_cash_usd'))
equity = checkpoint_number(report.get('ending_equity_usd'))
if report.get('calendar_year') != calendar_year or start != expected:
raise Blocked('YEAR_OR_START_BALANCE_MISMATCH')
if cash != equity or cash - start != income:
raise Blocked('PERIOD_RECONCILIATION_FAILED')
if checkpoint_number(outgoing['cash_usd']) != cash or checkpoint_number(outgoing['equity_usd']) != equity:
raise Blocked('CHECKPOINT_BALANCE_MISMATCH')
if checkpoint_number(report.get('cumulative_net_income_usd')) != cash - 500:
raise Blocked('CUMULATIVE_RECONCILIATION_FAILED')
if checkpoint_time(outgoing['as_of']).year != calendar_year:
raise Blocked('CHECKPOINT_YEAR_MISMATCH')
if incoming:
validate_checkpoint(incoming)
if checkpoint_time(outgoing['as_of']) <= checkpoint_time(incoming['as_of']):
raise Blocked('CHECKPOINT_TIME_REGRESSED')
for key in ('peak_equity_usd',):
if checkpoint_number(outgoing[key]) < checkpoint_number(incoming[key]):
raise Blocked('CHECKPOINT_RISK_STATE_REGRESSED')
if checkpoint_number(incoming['peak_equity_usd']) >= 1000 and checkpoint_number(outgoing['locked_floor_usd']) < 500:
raise Blocked('MILESTONE_FLOOR_LOWERED')
if incoming['halted'] and not outgoing['halted']:
raise Blocked('CHECKPOINT_HALT_CLEARED')
origin = report.get('data_origin')
if origin == 'SYNTHETIC_MACRO_SCENARIO':
if report.get('risk_scope') != 'MONTH_END_ONLY_NOT_INTRADAY':
raise Blocked('SYNTHETIC_RISK_SCOPE_REQUIRED')
if report.get('income_definition') != 'SYNTHETIC_REALIZED_NET_AFTER_MODELED_FEES':
raise Blocked('SYNTHETIC_INCOME_DEFINITION_REQUIRED')
elif origin in ('HISTORICAL_REPLAY', 'PARTIAL_REPLAY_NOT_CENTURY'):
# Original validator checks hashes, sources and historical risk claims.
# Actual capital and the original origin are validated/stored separately.
evidence = dict(report)
evidence['data_origin'] = 'HISTORICAL_REPLAY'
evidence['starting_capital_usd'] = '500'
validate_annual_evidence(evidence, calendar_year)
else:
raise Blocked('UNKNOWN_REPLAY_ORIGIN')
for key in ('strategy_sha256', 'dataset_sha256', 'ledger_sha256'):
value = report.get(key)
if not isinstance(value, str) or len(value) != 64 or any(c not in '0123456789abcdef' for c in value):
raise Blocked('EVIDENCE_HASH_REQUIRED')
return income, outgoing
def run_attempt(store, adapter, start_year, deadline=None, end_year=None):
as_of = datetime.now(timezone.utc)
end_year = as_of.year if end_year is None else end_year
if not 1926 <= start_year <= end_year <= as_of.year:
raise Blocked('INVALID_REPLAY_YEAR_RANGE')
attempt = store.start()
checkpoint = None
all_targets_met = True
origins = set()
store.event('ATTEMPT_STARTED', attempt=attempt, first_calendar_year=start_year,
last_calendar_year=end_year, years=end_year-start_year+1,
initial_capital_usd='500', capital_mode='CONTINUOUS_COMPOUNDING',
simulation_as_of=as_of.isoformat())
try:
for year, calendar_year in enumerate(range(start_year, end_year+1), 1):
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': calendar_year,
'starting_capital_usd': checkpoint['cash_usd'] if checkpoint else '500',
'account_checkpoint': checkpoint, 'income_target_usd': str(goal),
'tolerance_pct': '0.0001', 'lesson_history': store.lessons(), 'supervisor_log': store.snapshot()['events'],
'deadline_utc': deadline.isoformat() if deadline else None,
'simulation_as_of': as_of.isoformat(), 'mode': 'MIXED_SCENARIO_REPLAY',
'allow_live_orders': False, '_activity_db': str(store.path.resolve())}
store.event('YEAR_STARTED', attempt=attempt, year=year, calendar_year=calendar_year,
opening_cash_usd=context['starting_capital_usd'])
report = execute_year(adapter, context, deadline)
income, next_checkpoint = validate_compounded_result(report, calendar_year, checkpoint)
with store.lock, store.connect() as db:
db.execute('CREATE TABLE IF NOT EXISTS compounded_years (attempt INTEGER NOT NULL, calendar_year INTEGER NOT NULL, report_json TEXT NOT NULL, PRIMARY KEY(attempt,calendar_year))')
db.execute('INSERT INTO compounded_years VALUES(?,?,?)', (attempt, calendar_year, json.dumps(report)))
checkpoint = next_checkpoint
origins.add(report['data_origin'])
error = deviation(income, goal)
accepted = error <= Decimal('0.0001')
all_targets_met = all_targets_met and accepted
store.event('YEAR_RESULT', attempt=attempt, year=year, calendar_year=calendar_year,
data_origin=report['data_origin'], net_income_usd=str(income),
income_target_usd=str(goal), target_deviation_pct=str(error), accepted=accepted,
cash_usd=checkpoint['cash_usd'], equity_usd=checkpoint['equity_usd'],
account_checkpoint=checkpoint, missing_input_months=report.get('missing_input_months', []))
if calendar_year == 2019:
store.event('TRACK_HANDOFF_READY', attempt=attempt, from_year=2019, to_year=2020,
cash_usd=checkpoint['cash_usd'], equity_usd=checkpoint['equity_usd'])
if not accepted:
store.lesson('TARGET_DEVIATION_EXCEEDED',
{'income_usd': str(income), 'target_usd': str(goal), 'error_pct': str(error)},
'Inspect source coverage, decisions, risk constraints and modeled costs; test a specific change separately. No strategy weights were changed.', attempt, year)
store.event('EXPERIMENT_COMPLETE', attempt=attempt, origins=sorted(origins),
all_targets_met=all_targets_met, historical_century_validated=False,
strategy_weights_updated=False, account_checkpoint=checkpoint)
store.finish(attempt, 'COMPLETED_MIXED_SCENARIO')
return True
except Exception as exc:
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},
'Review the indicated input or strategy interface; preserve this attempt and its last committed checkpoint.', attempt)
store.finish(attempt, 'BLOCKED')
return False
ROOT=Path(__file__).resolve().parent
STOP=threading.Event()
"""
# END COMPLETE SOURCE: _finbot_supervisor
# BEGIN COMPLETE SOURCE: _finbot_adapter
COMPONENTS['_finbot_adapter'] = Path(__file__).with_name('host_replay_adapter.py').read_text(encoding='utf-8')
# END COMPLETE SOURCE: _finbot_adapter
def boot(project, config, strategy):
import types
loaded = {}
for name, source in COMPONENTS.items():
module = types.ModuleType(name)
module.__file__ = str(Path(__file__).resolve())
sys.modules[name] = module
exec(compile(source, __file__ + ':' + name, 'exec'), module.__dict__)
loaded[name] = module
supervisor = loaded['_finbot_supervisor']
adapter = loaded['_finbot_adapter']
supervisor.ROOT = project
adapter.ROOT = project
adapter.CONFIG = dict(config)
sys.modules['_finbot_selected_strategy'] = strategy
adapter.CONFIG['strategy_module'] = '_finbot_selected_strategy'
supervisor.execute_year = lambda adapter_module, context, deadline: adapter_module.run_year(context)
return supervisor, adapter
def load_config(project):
path = project / 'replay_config.json'
config = json.loads(path.read_text()) if path.is_file() else {}
if not isinstance(config, dict):
raise ValueError('replay_config.json must contain an object')
defaults = {'bars_path': 'data/replay_bars.jsonl',
'macro_path': 'data/macro_observations.jsonl',
'news_path': 'data/geopolitical_events.jsonl',
'output_path': 'runs/replay_ledgers',
'synthetic_macro_date_alias': True,
'synthetic_event_date_alias': True}
return {**defaults, **config}
def find_project(explicit=None):
if explicit:
project = Path(explicit).expanduser().resolve()
if not project.is_dir():
raise ValueError('Project directory does not exist')
return project
candidates = [Path('/Users/bpa/Desktop/ForesixaQuant/FinBot_Century_Gauntlet'),
Path(__file__).resolve().parent, Path.cwd()]
for candidate in candidates:
if (candidate / 'data' / 'macro_observations.jsonl').is_file():
return candidate.resolve()
raise ValueError('PROJECT_NOT_FOUND: place this file in the existing gauntlet project, or use --project')
def load_strategy(project, config, explicit=None):
import importlib.util
import ast
source = explicit or config.get('strategy_file')
if source:
source = Path(source).expanduser()
if not source.is_absolute():
source = project / source
elif config.get('strategy_module'):
name = config['strategy_module']
if not isinstance(name, str) or not all(part.isidentifier() for part in name.split('.')):
raise ValueError('INVALID_STRATEGY_MODULE')
source = project.joinpath(*name.split('.')).with_suffix('.py')
if not source.is_file():
package_file = project.joinpath(*name.split('.'), '__init__.py')
source = package_file if package_file.is_file() else source
else:
candidates = []
for folder in (project, project.parent):
for name in ('host_strategy.py', 'strategy.py'):
path = folder / name
if path.is_file():
tree = ast.parse(path.read_text(), filename=str(path))
names = {node.name for node in tree.body if isinstance(node, ast.FunctionDef)}
if {'decide', 'capital_floor'} <= names:
candidates.append(path)
if candidates:
break
if not candidates:
raise ValueError('HOST_STRATEGY_INTERFACE_MISSING: an actual decide(context) and capital_floor(context) strategy is required; no mock strategy was selected')
hashes = {hashlib.sha256(path.read_bytes()).hexdigest() for path in candidates}
if len(hashes) > 1:
raise ValueError('AMBIGUOUS_HOST_STRATEGY: preserve the existing strategy_module or strategy_file setting')
source = candidates[0]
if not source.is_file():
raise ValueError('HOST_STRATEGY_FILE_NOT_FOUND')
sys.path.insert(0, str(source.resolve().parent))
spec = importlib.util.spec_from_file_location('_finbot_selected_strategy', source.resolve())
module = importlib.util.module_from_spec(spec)
sys.modules[spec.name] = module
spec.loader.exec_module(module)
if not all(callable(getattr(module, key, None)) for key in ('decide', 'capital_floor')):
raise ValueError('HOST_STRATEGY_INTERFACE_MISSING')
return module
def select_db(project, explicit=None):
if explicit:
return Path(explicit).expanduser().resolve()
candidates = [project/'supervisor.db', project/'runs'/'gauntlet.db', project/'runs'/'supervisor.db']
existing = [path for path in candidates if path.is_file()]
if len(existing) > 1:
raise ValueError('MULTIPLE_REPLAY_DATABASES: use the existing --db argument; no ledger was replaced')
return existing[0] if existing else candidates[0]
def self_test():
import contextlib
import io
import tempfile
import types
from decimal import Decimal, localcontext
results = []
def passed(name):
results.append({'test': name, 'status': 'PASS'})
with tempfile.TemporaryDirectory(prefix='finbot-self-test-') as directory:
project = Path(directory)
data = project/'data'
data.mkdir()
macro = data/'macro_observations.jsonl'
news = data/'geopolitical_events.jsonl'
bars = data/'replay_bars.jsonl'
macro.write_text('\n'.join(json.dumps({
'available_at': f'2019-{month:02d}-01T00:00:00+00:00',
'series': 'INDPRO', 'observation_date': f'2019-{month:02d}-01',
'value': str(100+month), 'url': 'https://example.test/fixture-macro',
}) for month in range(1, 13)))
news.write_text('')
bars.write_text('\n'.join(json.dumps({
'at': f'2020-01-01T00:0{minute}:00+00:00',
'available_at': f'2020-01-01T00:0{minute}:00+00:00',
'data_origin': 'HISTORICAL_INTRADAY_BARS', 'bar_interval_seconds': 60,
'quotes': {'BTC/USD': str(100+minute), 'ETH/USD': str(10+minute)},
'source_url': 'https://example.test/fixture-bars',
}) for minute in (1, 2, 3)))
strategy = types.ModuleType('_fixture_strategy')
strategy.__file__ = __file__
def decide(context):
if context['mode'] == 'SYNTHETIC_MACRO_SCENARIO' and '-01-' in context['at']:
return [{'action': 'BUY', 'symbol': 'SYNTHETIC_MACRO_INDEX', 'quantity': '1'}]
return []
strategy.decide = decide
strategy.capital_floor = lambda context: context['locked_floor']
supervisor, adapter = boot(project, load_config(project), strategy)
store = supervisor.Store(project/'supervisor.db')
supervisor.STOP.clear()
with contextlib.redirect_stdout(io.StringIO()):
complete = supervisor.run_attempt(store, adapter, 1926, end_year=2025)
assert complete, 'Full fixture run did not complete'
with store.connect() as db:
reports = [json.loads(row[0]) for row in db.execute(
'SELECT report_json FROM compounded_years WHERE attempt=1 ORDER BY calendar_year')]
assert len(reports) == 100
assert reports[0]['calendar_year'] == 1926 and reports[-1]['calendar_year'] == 2025
assert db.execute('SELECT status FROM attempts WHERE id=1').fetchone()[0] == 'COMPLETED_MIXED_SCENARIO'
passed('complete_1926_through_2025_fixture_100_years')
with localcontext() as precision:
precision.prec = 4096
previous = Decimal(500)
for report in reports:
assert Decimal(report['starting_capital_usd']) == previous
cash = Decimal(report['ending_cash_usd'])
assert cash - previous == Decimal(report['net_income_usd'])
assert cash - 500 == Decimal(report['cumulative_net_income_usd'])
previous = cash
passed('all_100_period_and_cumulative_balances_reconcile')
assert reports[94]['calendar_year'] == 2020
assert reports[94]['starting_capital_usd'] == reports[93]['ending_cash_usd']
assert reports[94]['data_origin'] == 'PARTIAL_REPLAY_NOT_CENTURY'
assert reports[94]['account_checkpoint']['positions'] == []
passed('2019_to_2020_handoff_through_actual_embedded_execution_core')
assert reports[0]['data_origin'] == 'SYNTHETIC_MACRO_SCENARIO'
assert len(reports[0]['missing_input_months']) == 12
passed('missing_inputs_are_explicit_not_fabricated_history')
# Reopening preserves attempts and commits; this does not auto-resume a period.
reopened = supervisor.Store(project/'supervisor.db')
assert reopened.start() == 2
with contextlib.redirect_stdout(io.StringIO()):
reopened.finish(2, 'SELF_TEST_REOPENED')
passed('database_reopen_preserves_prior_attempts')
fixture_context = {'attempt': 3, 'year': 1, 'calendar_year': 2019,
'starting_capital_usd': '500', 'allow_live_orders': False}
news.write_text(json.dumps({'available_at': '2019-02-01T00:00:00+00:00',
'event_id': 'fixture-shock', 'goldstein': '-9', 'tone': '-15'}))
with contextlib.redirect_stdout(io.StringIO()):
report = adapter.run_year(fixture_context)
assert report['account_checkpoint']['halted']
assert report['account_checkpoint']['positions'] == []
passed('shock_liquidation_and_persistent_halt')
strategy.capital_floor = lambda context: '-1'
try:
adapter.run_year(fixture_context)
except supervisor.Blocked as exc:
assert str(exc) == 'INVALID_HOST_CAPITAL_FLOOR'
else:
raise AssertionError('Negative floor accepted')
passed('invalid_capital_floor_rejected')
strategy.capital_floor = lambda context: context['locked_floor']
broken = json.loads(json.dumps(reports[94]))
broken['starting_capital_usd'] = '500'
try:
supervisor.validate_compounded_result(broken, 2020, reports[93]['account_checkpoint'])
except supervisor.Blocked:
pass
else:
raise AssertionError('Wrong rollover balance accepted')
passed('wrong_opening_balance_rejected')
future_rows = adapter.read_scenario_inputs(macro)
assert all(at.year == 2019 for at, row in future_rows)
prior_year = list(adapter.synthetic_macro_contexts(1926))
assert all(row['input_status'] == 'NO_INPUTS' for row in prior_year)
passed('future_macro_releases_not_visible_in_1926')
# Current-year synthesis must not emit a future month-end.
cutoff = '2026-03-15T12:00:00+00:00'
rows = list(adapter.synthetic_macro_contexts(2026, cutoff))
assert len(rows) == 3 and rows[-1]['at'] == cutoff
passed('current_year_stops_at_observation_cutoff')
# A stop is recorded without completing another simulated year.
supervisor.STOP.set()
with contextlib.redirect_stdout(io.StringIO()):
assert not supervisor.run_attempt(store, adapter, 1926, end_year=2025)
supervisor.STOP.clear()
passed('supervisor_stop_is_respected')
print(json.dumps({'status': 'PASS', 'tests': results,
'scope': 'One-file synthetic/fixture replay only; no private host strategy, real dataset completeness, live trading or Docker certification.'}, indent=2))
return 0
def make_server(store, port=8891, dashboard=None):
"""Read-only loopback analysis service. Disconnects do not stop replay."""
import socket, threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import urlsplit, parse_qs
disconnected=(BrokenPipeError,ConnectionResetError,ConnectionAbortedError,TimeoutError,socket.timeout)
slots=threading.BoundedSemaphore(16)
page=b'''FinBot activityFinBot replay activity
Latest 200 displayed; all history remains in the ledger.
'''
class Server(ThreadingHTTPServer):
daemon_threads=True
block_on_close=False
def process_request(self,request,address):
if not slots.acquire(False):request.close();return
try:super().process_request(request,address)
except BaseException:slots.release();raise
def process_request_thread(self,request,address):
try:super().process_request_thread(request,address)
finally:slots.release()
class Handler(BaseHTTPRequestHandler):
def setup(self):
super().setup();self.connection.settimeout(10)
def log_message(self,*args):pass
def handle(self):
try:super().handle()
except disconnected:pass
def finish(self):
try:super().finish()
except disconnected:pass
def response(self,value,kind='application/json',code=200):
body=value if isinstance(value,bytes) else json.dumps(value).encode()
self.send_response(code);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)
def do_GET(self):
host=self.headers.get('Host','').split(':')[0]
if host not in ('127.0.0.1','localhost'):self.response({'error':'Local service only'},code=403);return
parsed=urlsplit(self.path);query=parse_qs(parsed.query)
try:
if parsed.path=='/':
self.response(Path(dashboard).read_bytes() if dashboard else page,'text/html; charset=utf-8')
elif parsed.path=='/api/activity':
self.response(store.activity(after=int(query.get('after',['0'])[0]),before=int(query['before'][0]) if 'before' in query else None,q=query.get('q',[''])[0],kind=query.get('kind',[''])[0],limit=int(query.get('limit',['100'])[0])))
elif parsed.path=='/api/status':self.response(store.snapshot())
elif parsed.path=='/api/lessons':self.response(store.activity(kind='LESSON',after=int(query.get('after',['0'])[0]),limit=int(query.get('limit',['100'])[0])))
elif parsed.path=='/api/export':
self.send_response(200);self.send_header('Content-Type','application/x-ndjson');self.send_header('Cache-Control','no-store');self.end_headers()
with store.connect() as db:
maximum=db.execute('SELECT COALESCE(max(id),0) FROM events').fetchone()[0]
cursor=db.execute('SELECT id,created_at,kind,payload FROM events WHERE id<=? ORDER BY id',(maximum,))
while True:
batch=cursor.fetchmany(100)
if not batch:break
for i,at,kind,payload in batch:self.wfile.write((json.dumps(dict(id=i,created_at=at,kind=kind,**json.loads(payload)))+'\n').encode())
elif parsed.path=='/api/stream':
self.send_response(200);self.send_header('Content-Type','text/event-stream');self.send_header('Cache-Control','no-store');self.end_headers()
after=int(query.get('after',['0'])[0]);end=__import__('time').monotonic()+120
while __import__('time').monotonic()