"""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_at