"""Historical-clock bridge to the recovered FinBot execution/risk engine. No broker imports or order APIs. Entry decisions come from the host strategy, not from target income or invented research. See REPLAY_INPUTS.txt. """ from contextlib import contextmanager from datetime import datetime,timezone from decimal import Decimal import hashlib import importlib import json import os from pathlib import Path from gauntlet import Blocked from vendor import finbot_engine as core ROOT=Path(__file__).resolve().parent def instant(value): value=datetime.fromisoformat(value.replace('Z','+00:00')) if value.tzinfo is None:raise Blocked('TIMEZONE_REQUIRED') return value.astimezone(timezone.utc) def digest(path): h=hashlib.sha256() with Path(path).open('rb') as f: for block in iter(lambda:f.read(1024*1024),b''):h.update(block) return h.hexdigest() @contextmanager def historical_clock(cursor): original_datetime,original_timestamp=core.datetime,core.timestamp class ReplayDateTime(datetime): @classmethod def now(cls,tz=None):return cursor[0].astimezone(tz) if tz else cursor[0].replace(tzinfo=None) core.datetime=ReplayDateTime;core.timestamp=lambda:cursor[0].isoformat() old={k:os.environ.get(k) for k in ('TRADING_MODE','ALPACA_BASE_URL')} os.environ['TRADING_MODE']='LOCAL_SIMULATION';os.environ['ALPACA_BASE_URL']='https://paper-api.alpaca.markets' try:yield finally: core.datetime,core.timestamp=original_datetime,original_timestamp for k,v in old.items(): if v is None:os.environ.pop(k,None) else:os.environ[k]=v def macro_snapshots(path): snapshots={} with Path(path).open() as stream: for line in stream: r=json.loads(line);available=instant(r['available_at']);series=r['series'] snapshot=snapshots.setdefault(available,{}) if series not in snapshot or snapshot[series]['observation_date']at:break values.update(rows) return values def year_bars(path,year): previous=None;count=0 with Path(path).open() as stream: for line in stream: r=json.loads(line);at=instant(r['at']) if previous is not None and at<=previous:raise Blocked('BARS_NOT_CHRONOLOGICAL') previous=at if at.year!=year:continue if r.get('data_origin') not in ('HISTORICAL_INTRADAY_QUOTES','HISTORICAL_INTRADAY_BARS'):raise Blocked('VERIFIED_INTRADAY_QUOTES_REQUIRED') seconds=r.get('bar_interval_seconds') if type(seconds) is not int or not 0at:raise Blocked('BAR_LOOKAHEAD') if not r.get('quotes') or not isinstance(r['quotes'],dict):raise Blocked('QUOTES_REQUIRED') if not isinstance(r.get('source_url'),str) or not r['source_url'].startswith('https://'):raise Blocked('BAR_SOURCE_REQUIRED') if any(symbol not in ('BTC/USD','ETH/USD') for symbol in r['quotes']):raise Blocked('RECOVERED_ENGINE_UNIVERSE_MISMATCH') for price in r['quotes'].values(): if not isinstance(price,str):raise Blocked('EXACT_PRICE_STRINGS_REQUIRED') core.number(price,True) count+=1;yield at,r if not count:raise Blocked('NO_MARKET_BARS_FOR_YEAR:'+str(year)) def validate_research(r,at): if not isinstance(r,dict):raise Blocked('HOST_RESEARCH_REQUIRED') for channel in core.CHANNELS: for source in r.get(channel,{}).get('sources',[]): if 'available_at' not in source or instant(source['available_at'])>at:raise Blocked('RESEARCH_LOOKAHEAD_OR_UNKNOWN_PUBLICATION') if instant(source['observed_at'])>at:raise Blocked('RESEARCH_OBSERVED_IN_FUTURE') def replay(context,strategy,bars_path,macro_path,output_dir,require_full_year=True): year=context['calendar_year'];bars=iter(year_bars(bars_path,year)) try:first=next(bars) except StopIteration:raise Blocked('NO_MARKET_BARS_FOR_YEAR:'+str(year)) # BTC/ETH are not silently relabeled into a fictional century of prices. if year<2009:raise Blocked('RECOVERED_CRYPTO_ENGINE_HAS_NO_1926_UNIVERSE') if require_full_year and first[0] > datetime(year,1,1,0,1,tzinfo=timezone.utc):raise Blocked('YEAR_START_COVERAGE_MISSING') snapshots=macro_snapshots(macro_path);cursor=[first[0]];sources={};previous_at=None from activity_privacy import clean from gauntlet import Store live=Store(context['_activity_db']) if context.get('_activity_db') else None class ActivityList(list): def append(self,row): row=clean(row);super().append(row) if live:live.event('REPLAY_'+row['type'],attempt=context['attempt'],year=context['year'] if 'year' in context else year,calendar_year=year,activity=row) events=ActivityList();seen_news=set() with historical_clock(cursor): engine=core.Engine(':memory:');engine.initialize('500');floor=Decimal(0);peak=Decimal(500) try: import itertools for index,(at,row) in enumerate(itertools.chain([first],bars)): if require_full_year and previous_at is not None and (at-previous_at).total_seconds()>row['bar_interval_seconds']:raise Blocked('INTRADAY_COVERAGE_GAP') previous_at=at;cursor[0]=at if context.get('deadline_utc') and datetime.now(timezone.utc)>=instant(context['deadline_utc']):raise Blocked('REPLAY_DEADLINE') quotes={s:{'price':p,'at':at.isoformat()} for s,p in row['quotes'].items()} if any(p['symbol'] not in quotes for p in engine.positions()):raise Blocked('MISSING_OPEN_POSITION_MARK') events.append({'type':'MARKET_DATA','at':at.isoformat(),'quotes':row['quotes'],'source_url':row['source_url']}) marks=engine.mark(quotes) eq=engine.equity();peak=max(peak,eq) events.append({'type':'RISK_CHECK','at':at.isoformat(),'drawdown':marks['daily_drawdown'],'trailing_drawdown':marks['trailing_drawdown'],'halted':bool(engine.account()['halted']),'protective_exits':marks['exits'],'stop_updates':marks['stop_updates']}) next_floor=core.number(strategy.capital_floor({'equity':str(eq),'peak_equity':str(peak),'locked_floor':str(floor),'starting_capital_usd':'500','at':at.isoformat()})) if next_floorpeak:raise Blocked('INVALID_HOST_CAPITAL_FLOOR') if next_floor!=floor:events.append({'type':'FLOOR_RATCHET','at':at.isoformat(),'floor':str(next_floor)}) events.append({'type':'FLOOR_CHECK','at':at.isoformat(),'previous_floor':str(floor),'next_floor':str(next_floor),'equity':str(eq)}) floor=next_floor if floor and eq<=floor: with engine.transaction(): for p in list(engine.positions()):engine.close_position(p['symbol'],quotes[p['symbol']]['price'],at.isoformat(),'CAPITAL_FLOOR_TRIGGER') engine.db.execute('UPDATE accounts SET halted=1 WHERE id=1') events.append({'type':'FLOOR_TRIGGERED','at':at.isoformat(),'observed_equity':str(engine.equity()),'floor':str(floor),'floor_is_not_guaranteed':True}) macro=macro_at(snapshots,at) from geopolitical_news import available_news news=available_news(ROOT/'data/geopolitical_events.jsonl',at.isoformat()) for item in news['events']: if item['event_id'] not in seen_news: seen_news.add(item['event_id']);events.append({'type':'GEOPOLITICAL_NEWS','at':at.isoformat(),'news':{k:v for k,v in item.items() if k!='raw_fields'}}) events.append({'type':'CONTEXT','at':at.isoformat(),'macro':macro,'geopolitical':{k:v for k,v in news.items() if k!='events'}}) events.append({'type':'EVALUATION_STARTED','at':at.isoformat(),'account':engine.status()}) # Do not expose targets to the strategy as if they were market evidence. decisions=strategy.decide({'at':at.isoformat(),'quotes':row['quotes'],'macro':macro,'geopolitical':news, 'account':engine.status(),'lesson_history':context.get('lesson_history',[]),'mode':'HISTORICAL_REPLAY_ONLY'}) if not isinstance(decisions,list):raise Blocked('STRATEGY_DECISIONS_MUST_BE_ARRAY') if not decisions:events.append({'type':'DECISION','at':at.isoformat(),'action':'NO_INSTRUCTION','reason':'Strategy returned no instructions.'}) for d in decisions: events.append({'type':'DECISION','at':at.isoformat(),'decision':d}) action=d.get('action');symbol=d.get('symbol') if action=='HOLD':continue if symbol not in quotes:raise Blocked('DECISION_WITHOUT_QUOTE') try: if action=='BUY': validate_research(d.get('research'),at) events.append({'type':'ORDER_SUBMITTED','at':at.isoformat(),'action':action,'symbol':symbol,'mode':'SIMULATED'}) result=engine.simulate_entry(f"a{context['attempt']}-y{year}-{index}-{symbol}",symbol,quotes[symbol]['price'],d['stop'],at.isoformat(),d['research'],requested_qty=d.get('quantity'),trailing_pct=d.get('trailing_pct'),execution_tag='HISTORICAL_REPLAY' if d.get('trailing_pct') else None) elif action=='SELL': events.append({'type':'ORDER_SUBMITTED','at':at.isoformat(),'action':action,'symbol':symbol,'mode':'SIMULATED'}) with engine.transaction():result=engine.close_position(symbol,quotes[symbol]['price'],at.isoformat(),'HOST_STRATEGY_EXIT') else:raise Blocked('UNKNOWN_STRATEGY_ACTION') events.append({'type':'EXECUTION','at':at.isoformat(),'result':result}) except core.Rejected as exc: events.append({'type':'ENGINE_REJECTION','at':at.isoformat(),'reason':str(exc)}) sources[row['source_url']]={'url':row['source_url'],'point_in_time_validated':True,'timing_basis':'Host supplied bar provenance; not externally authenticated.'} for v in macro.values():sources[v['url']]={'url':v['url'],'point_in_time_validated':True,'timing_basis':'Verified ALFRED vintage header plus conservative snapshot availability.'} events.append({'type':'MARK','at':at.isoformat(),'result':marks,'equity':str(engine.equity()),'locked_floor':str(floor),'positions':[dict(p) for p in engine.positions()]}) if require_full_year and cursor[0] < datetime(year,12,31,23,59,tzinfo=timezone.utc):raise Blocked('YEAR_END_COVERAGE_MISSING') # Close remaining inventory for consistent realized annual income. with engine.transaction(): for p in list(engine.positions()): result=engine.close_position(p['symbol'],p['mark'],cursor[0].isoformat(),'REPLAY_YEAR_END') events.append({'type':'EXECUTION','at':cursor[0].isoformat(),'cause':'REPLAY_YEAR_END','result':result}) rows=[dict(r) for r in engine.db.execute('SELECT * FROM dispositions ORDER BY id')] income=sum((Decimal(r['economic_pnl']) for r in rows),Decimal(0)) output_dir=Path(output_dir);output_dir.mkdir(parents=True,exist_ok=True) ledger=output_dir/f"attempt_{context['attempt']}_year_{year}.json" ledger.write_text(json.dumps({'dispositions':rows,'events':events,'final_cash':str(engine.equity()),'measurement':'SIMULATED_EXECUTION_ON_HOST_SUPPLIED_HISTORICAL_QUOTES'},indent=2)) if engine.equity()-Decimal(500)!=income:raise Blocked('LEDGER_RECONCILIATION_FAILED') return {'data_origin':'HISTORICAL_REPLAY' if require_full_year else 'PARTIAL_REPLAY_NOT_CENTURY','calendar_year':year,'starting_capital_usd':'500', 'net_income_usd':str(income),'income_definition':'REALIZED_NET_AFTER_COSTS', 'strategy_sha256':digest(strategy.__file__),'dataset_sha256':hashlib.sha256((digest(bars_path)+digest(macro_path)).encode()).hexdigest(),'ledger_sha256':digest(ledger), 'sources':list(sources.values()),'controls':dict.fromkeys(('intraday_3pct_breaker','capital_floor_ratchet','no_live_orders','point_in_time_inputs'),True), 'control_scope':'Evaluated at supplied observations; unseen intrabar moves and floor gaps are not guaranteed.'} finally:engine.db.close() def run_year(context): config=json.loads((ROOT/'replay_config.json').read_text()) bars=ROOT/config['bars_path'];macro=ROOT/config['macro_path'] if not bars.is_file():raise Blocked('MARKET_REPLAY_BARS_NOT_INSTALLED') if not macro.is_file():raise Blocked('MACRO_SNAPSHOTS_NOT_INSTALLED') name=config.get('strategy_module') if not name:raise Blocked('HOST_STRATEGY_MODULE_NOT_CONFIGURED') strategy=importlib.import_module(name) if not all(callable(getattr(strategy,k,None)) for k in ('decide','capital_floor')):raise Blocked('HOST_STRATEGY_INTERFACE_MISSING') return replay(context,strategy,bars,macro,ROOT/config['output_path'])