"""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, localcontext import calendar import hashlib import importlib import json import os from pathlib import Path from _finbot_supervisor import Blocked import _finbot_core as core ROOT=Path(__file__).resolve().parent CONFIG={} def cash_number(value): if isinstance(value,bool):raise Blocked('INVALID_DECIMAL') try:n=Decimal(str(value)) except Exception:raise Blocked('INVALID_DECIMAL') if not n.is_finite():raise Blocked('NONFINITE_DECIMAL') return n def checked_checkpoint(context): cp=context.get('account_checkpoint') if cp is None: if cash_number(context.get('starting_capital_usd','500'))!=500:raise Blocked('INITIAL_CAPITAL_MUST_BE_500') return None if cp.get('positions')!=[] or type(cp.get('halted')) is not bool:raise Blocked('INVALID_CHECKPOINT') cash=cash_number(cp['cash_usd']);equity=cash_number(cp['equity_usd']) floor=cash_number(cp['locked_floor_usd']);peak=cash_number(cp['peak_equity_usd']) if cash<0 or cash!=equity or not 0<=floor<=peak or peak=first_at:raise Blocked('CHECKPOINT_TIME_INVALID') engine.initialize(str(cash)) if cp: with engine.transaction(): engine.db.execute('UPDATE accounts SET initial_equity=?,session=?,session_equity=?,peak_equity=?,halted=? WHERE id=1', ('500',cp['session'],cp['session_equity_usd'],cp['peak_equity_usd'],int(cp['halted']))) return cash,cash_number(cp['locked_floor_usd']) if cp else Decimal(0),cash_number(cp['peak_equity_usd']) if cp else cash def account_checkpoint(engine,at,floor,peak,context): a=engine.account();cash=cash_number(a['cash']) if list(engine.positions()) or cash!=engine.equity():raise Blocked('CLOSED_ACCOUNT_REQUIRED') cp={'version':1,'as_of':at.isoformat(),'cash_usd':str(cash),'equity_usd':str(cash),'positions':[], 'locked_floor_usd':str(floor),'peak_equity_usd':str(max(peak,cash,cash_number(a['peak_equity']))), 'initial_capital_usd':'500','cumulative_net_income_usd':str(cash-500),'halted':bool(a['halted']), 'session':at.date().isoformat(),'session_equity_usd':a['session_equity'] if a['session']==at.date().isoformat() else str(cash)} for key in ('synthetic_index_usd','synthetic_macro_state'): if key in (context.get('account_checkpoint') or {}):cp[key]=context['account_checkpoint'][key] return cp 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): return DiskRows(path) def macro_at(snapshots,at): values={} for available,row in snapshots: if available>at:break series=row['series'];old=values.get(series) if old is None or row['observation_date']>=old['observation_date']:values[series]=row 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 crypto_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 _finbot_privacy import clean from _finbot_supervisor 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), localcontext() as precision: precision.prec=4096 engine=core.Engine(':memory:') try: starting_cash,floor,peak=restore_account(engine,context,first[0]) 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':str(starting_cash),'initial_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 _finbot_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)) checkpoint=account_checkpoint(engine,cursor[0],floor,peak,context) 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,'starting_cash_usd':str(starting_cash),'net_income_usd':str(income),'account_checkpoint':checkpoint,'final_cash':str(engine.equity()),'measurement':'SIMULATED_EXECUTION_ON_HOST_SUPPLIED_HISTORICAL_QUOTES'},indent=2)) if engine.equity()-starting_cash!=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':str(starting_cash), 'ending_cash_usd':checkpoint['cash_usd'],'ending_equity_usd':checkpoint['equity_usd'], 'cumulative_net_income_usd':checkpoint['cumulative_net_income_usd'],'account_checkpoint':checkpoint, 'full_year_coverage':require_full_year, '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() class DiskRows: """Disk-sorted JSONL sequence; source bytes are never collected into a list. Temporary SQLite is owned by this instance and removed on close/GC. Invalid rows are reported without exposing their content. """ def __init__(self,path,alias_macro=False,alias_events=False): import sqlite3,tempfile self.temp=tempfile.TemporaryDirectory(prefix='finbot-input-') self.db=sqlite3.connect(str(Path(self.temp.name)/'index.sqlite')) self.db.execute('PRAGMA cache_size=-2048') self.db.execute('CREATE TABLE raw(at TEXT,payload TEXT)') with Path(path).open(encoding='utf-8') as f: for lineno,line in enumerate(f,1): if not line.strip():continue try: row=json.loads(line) if row.get('retrospective_only') is True:continue at=instant(row['available_at']).isoformat() if alias_macro and str(row.get('observation_date',''))[:4].isdigit() and int(str(row['observation_date'])[:4])<2000: observation=str(row['observation_date']) assumed=instant(observation+'T00:00:00+00:00' if len(observation)==10 else observation) row['source_available_at']=row['available_at'] row['scenario_available_at']=assumed.isoformat() row['availability_basis']='ASSUMED_OBSERVATION_DATE_NOT_HISTORICAL_PUBLICATION' row['point_in_time_validated']=False row['available_at']=assumed.isoformat() at=assumed.isoformat() if alias_events: event_date=next((row.get(k) for k in ('event_date','event_at','at','SQLDATE') if row.get(k) not in (None,'')),None) if event_date is not None: event_date=str(event_date) if len(event_date)==8 and event_date.isdigit():event_date=event_date[:4]+'-'+event_date[4:6]+'-'+event_date[6:] event_time=instant(event_date+'T00:00:00+00:00' if len(event_date)==10 else event_date) if event_time.year<2000: row['source_available_at']=row.get('source_available_at',row['available_at']) row['scenario_available_at']=event_time.isoformat() row['available_at']=event_time.isoformat() row['availability_basis']='ASSUMED_EVENT_DATE_NOT_HISTORICAL_PUBLICATION' row['point_in_time_validated']=False at=event_time.isoformat() self.db.execute('INSERT INTO raw VALUES(?,?)',(at,json.dumps(row))) except (ValueError,KeyError,TypeError): print(json.dumps({'kind':'DATA_FAULT','file':Path(path).name,'line':lineno,'action':'SKIPPED_INVALID_ROW'}),flush=True) self.db.execute('CREATE TABLE ordered AS SELECT row_number() OVER(ORDER BY at,rowid)-1 AS pos,at,payload FROM raw') self.db.execute('CREATE UNIQUE INDEX by_pos ON ordered(pos)') self.db.execute('DROP TABLE raw');self.db.commit() self.count=self.db.execute('SELECT count(*) FROM ordered').fetchone()[0] def __len__(self):return self.count def __getitem__(self,n): if not isinstance(n,int):raise TypeError('integer index required') if n<0:n+=self.count row=self.db.execute('SELECT at,payload FROM ordered WHERE pos=?',(n,)).fetchone() if row is None:raise IndexError(n) return instant(row[0]),json.loads(row[1]) def __iter__(self): for at,payload in self.db.execute('SELECT at,payload FROM ordered ORDER BY pos'): yield instant(at),json.loads(payload) def __del__(self): try:self.db.close();self.temp.cleanup() except Exception:pass def read_scenario_inputs(path,alias_macro=False,alias_events=False): return DiskRows(path,alias_macro=alias_macro,alias_events=alias_events) def synthetic_macro_contexts(year,cutoff=None): config=dict(CONFIG) macro_rows=read_scenario_inputs(ROOT/config['macro_path'],alias_macro=year<2000 and config.get('synthetic_macro_date_alias',True)) news_rows=read_scenario_inputs(ROOT/config.get('news_path','data/geopolitical_events.jsonl'),alias_events=year<2000 and config.get('synthetic_event_date_alias',True)) cutoff=instant(cutoff) if cutoff else datetime.now(timezone.utc) first=datetime(year,1,1,tzinfo=timezone.utc) latest={};mi=ni=0;index=Decimal(100) def ingest(row): if row.get('value') in (None,'','.'):return old=latest.get(row['series']) if old is None or row['observation_date']>=old['observation_date']:latest[row['series']]=row while micutoff:break at=min(cutoff,datetime(year,month,calendar.monthrange(year,month)[1],23,59,59,tzinfo=timezone.utc)) while mi0:r+=Decimal(weight)*(values[series]/previous[series]-1) for series,weight in (('UNRATE','-.01'),('FEDFUNDS','-.005')): if series in values and series in previous:r+=Decimal(weight)*(values[series]-previous[series]) scores=[cash_number(v['goldstein']) for v in news if v.get('goldstein') not in (None,'','.')] tones=[cash_number(v['tone']) for v in news if v.get('tone') not in (None,'','.')] shock=bool((scores and sum(scores)/len(scores)<=-5) or (tones and sum(tones)/len(tones)<=-10)) if shock:r-=Decimal('.10') r=max(Decimal('-.50'),min(Decimal('.50'),r)) index=max(Decimal('.000000000001'),(index*(1+r)).quantize(Decimal('.000000000001'))) previous=values yield {'at':at.isoformat(),'quotes':{'SYNTHETIC_MACRO_INDEX':str(index)},'macro':dict(latest), 'geopolitical':{'events':news,'shock':shock,'tone':str(sum(tones)/len(tones)) if tones else '0','goldstein':str(sum(scores)/len(scores)) if scores else '0','status':'OWNER_SUPPLIED_SCENARIO_EVENTS' if news else 'ASSUMED_NEUTRAL_NO_EVENTS','is_assumed':not bool(news) or any(v.get('point_in_time_validated') is False for v in news)},'model_return':str(r), 'macro_state':{k:str(v) for k,v in values.items()}, 'input_status':'INPUTS_AVAILABLE' if latest or news else 'NO_INPUTS', 'data_origin':'SYNTHETIC_MACRO_SCENARIO'} def replay(context,strategy,bars_path,macro_path,output_dir,require_full_year=True): if not 1926<=context['calendar_year']<=2019: return crypto_replay(context,strategy,bars_path,macro_path,output_dir,require_full_year) with localcontext() as precision: precision.prec=4096 return synthetic_replay(context,strategy,output_dir) def synthetic_replay(context,strategy,output_dir): year=context['calendar_year'];cp=checked_checkpoint(context) or {} cash=cash_number(cp.get('cash_usd','500'));starting_cash=cash units=Decimal(0);price=cash_number(cp.get('synthetic_index_usd','100')) peak=cash_number(cp.get('peak_equity_usd','500'));floor=cash_number(cp.get('locked_floor_usd','250')) committed=cash_number(cp.get('bootstrap_purchase_cost_usd','0')) milestone=bool(cp.get('bootstrap_milestone',False) or peak>=1000) halted=bool(cp.get('halted',False));events=ActivityEvents(context);missing_months=[];last_observation=None if price<=0:raise Blocked('INVALID_SYNTHETIC_INDEX') money_tick=Decimal('.00000001');price_tick=Decimal('.000000000001');fee_rate=Decimal('.0025') def fill(side,quantity,at,reason): nonlocal cash,units,floor,committed,milestone if quantity<0 or quantity!=quantity.quantize(money_tick):raise Blocked('INVALID_SYNTHETIC_QUANTITY') gross=(quantity*price).quantize(money_tick);fee=(gross*fee_rate).quantize(money_tick) if side=='BUY': if gross+fee>cash:raise Blocked('INSUFFICIENT_SYNTHETIC_CASH') proposed_committed=committed+gross+fee proposed_floor=Decimal(500) if milestone else Decimal(500)-proposed_committed if proposed_floor < Decimal(250): events.append({"type":"ORDER_REJECTED","at":at,"reason":"PURCHASE_ALLOWANCE_EXCEEDED","proposed_floor_usd":str(proposed_floor),"available_purchase_allowance_usd":str(max(Decimal(0),Decimal(250)-committed))}) return if cash+units*price-fee < proposed_floor:raise Blocked('PURCHASE_WOULD_BREACH_FLOOR') cash-=gross+fee;units+=quantity;committed=proposed_committed;floor=proposed_floor elif side=='SELL': if quantity>units:raise Blocked('INSUFFICIENT_SYNTHETIC_UNITS') cash+=gross-fee;units-=quantity else:raise Blocked('UNKNOWN_SYNTHETIC_ACTION') events.append({'type':'FILL','at':at,'side':side,'quantity':str(quantity),'price':str(price), 'fee_usd':str(fee),'cash_usd':str(cash),'reason':reason}) for observation in synthetic_macro_contexts(year,context.get('simulation_as_of')): deadline=context.get('deadline_utc') if deadline and datetime.now(timezone.utc)>=instant(deadline):raise Blocked('REPLAY_DEADLINE') last_observation=observation;at=observation['at'];equity_before_mark=cash+units*price price=max(price_tick,(price*(1+cash_number(observation['model_return']))).quantize(price_tick)) equity=cash+units*price;peak=max(peak,equity) reported_floor = cash_number(strategy.capital_floor({"equity":str(equity), "peak_equity":str(peak), "locked_floor":str(floor), "starting_capital_usd":str(starting_cash), "initial_capital_usd":"500", "at":at, "mode":"SYNTHETIC_MACRO_SCENARIO"})) if reported_floor < 0 or reported_floor > peak:raise Blocked("INVALID_HOST_CAPITAL_FLOOR") if equity >= Decimal(1000): milestone=True floor=Decimal(500) observation_drawdown = ( (equity_before_mark - equity) / equity_before_mark if equity_before_mark > 0 else Decimal("0") ) halted = bool( halted or observation["geopolitical"]["shock"] or (floor > 0 and equity <= floor) or observation_drawdown >= Decimal(".03") or (peak - equity) / peak >= Decimal(".10") ) if observation["input_status"] == "NO_INPUTS":missing_months.append(at) decisions = [] if halted: if units:fill("SELL", units, at, "RISK_HALT") else: strategy_context = { **observation, "quotes": {"SYNTHETIC_MACRO_INDEX": str(price)}, "account": {"cash_usd": str(cash),"equity_usd": str(equity),"units": str(units)}, "lesson_history": context.get("lesson_history", []), "supervisor_log": context.get("supervisor_log", []), "fills": [dict(event) for event in events if event.get("type") == "FILL"], "learning_state": context.get("learning_state", {}), "learning_state_status": "HOST_MANAGED_NOT_FRAMEWORK_TRAINED", "risk": { "cash_usd": str(cash), "equity_usd": str(equity), "floor_usd": str(floor), "minimum_floor_usd": "250", "headroom_usd": str(max(Decimal(0), equity-floor)), "purchase_cost_total_usd": str(committed), "milestone_locked": milestone, "halted": halted, "fee_rate": str(fee_rate), "max_cash_purchase_notional_usd": str(cash/(1+fee_rate)), "requires_execution_revalidation": True, }, "mode": "SYNTHETIC_MACRO_SCENARIO","allow_live_orders": False, } decisions = strategy.decide(strategy_context) if not isinstance(decisions, list):raise Blocked("STRATEGY_DECISIONS_MUST_BE_ARRAY") for decision in decisions: if not isinstance(decision, dict):raise Blocked("INVALID_SYNTHETIC_INSTRUCTION") action = decision.get("action") if action == "HOLD":continue if decision.get("symbol") != "SYNTHETIC_MACRO_INDEX":raise Blocked("SYNTHETIC_STRATEGY_SYMBOL_REQUIRED") fill(action,cash_number(decision["quantity"]),at,"HOST_STRATEGY") events.append({"type":"MONTH","at":at,"synthetic_price":str(price),"cash_usd":str(cash), "equity_usd":str(cash+units*price),"locked_floor_usd":str(floor),"halted":halted, "input_status":observation["input_status"],"decisions":decisions}) if last_observation is None:raise Blocked("NO_SYNTHETIC_OBSERVATIONS") at=last_observation['at'] if units:fill('SELL',units,at,'YEAR_END_LIQUIDATION') income=cash-starting_cash outgoing={'version':1,'as_of':at,'cash_usd':str(cash),'equity_usd':str(cash),'positions':[], 'locked_floor_usd':str(floor),'peak_equity_usd':str(max(peak,cash)),'initial_capital_usd':'500', 'cumulative_net_income_usd':str(cash-500),'halted':halted,'session':instant(at).date().isoformat(), 'session_equity_usd':str(cash if cash>0 else starting_cash),'synthetic_index_usd':str(price), 'synthetic_macro_state':last_observation['macro_state'], 'bootstrap_purchase_cost_usd':str(committed), 'bootstrap_milestone':milestone} report={'data_origin':'SYNTHETIC_MACRO_SCENARIO','calendar_year':year,'starting_capital_usd':str(starting_cash), 'net_income_usd':str(income),'ending_cash_usd':str(cash),'ending_equity_usd':str(cash), 'cumulative_net_income_usd':outgoing['cumulative_net_income_usd'], 'income_definition':'SYNTHETIC_REALIZED_NET_AFTER_MODELED_FEES','account_checkpoint':outgoing, 'full_year_coverage':False,'missing_input_months':missing_months,'risk_scope':'MONTH_END_ONLY_NOT_INTRADAY','macro_date_alias':year<2000 and CONFIG.get('synthetic_macro_date_alias',True),'news_assumption':'OWNER_SUPPLIED_EVENT_DATE_ALIAS_OR_NEUTRAL_IF_MISSING' if year<2000 else 'SOURCE_AVAILABILITY', 'runway_phase':'A_1926_1978' if year<=1978 else 'B_1979_ONWARD'} folder=Path(output_dir);folder.mkdir(parents=True,exist_ok=True) ledger=folder/f"attempt_{context['attempt']}_year_{year}_synthetic.json" ledger.write_text(json.dumps({'report':report,'events':events},indent=2),encoding='utf-8') config=dict(CONFIG) report['ledger_sha256']=digest(ledger);report['strategy_sha256']=digest(strategy.__file__) report['dataset_sha256']=hashlib.sha256((digest(ROOT/config['macro_path'])+digest(ROOT/config.get('news_path','data/geopolitical_events.jsonl'))).encode()).hexdigest() return report def run_year(context): config=dict(CONFIG) name=config.get('strategy_module') if not isinstance(name,str) or not name.strip():raise Blocked('HOST_STRATEGY_MODULE_NOT_CONFIGURED') strategy=importlib.import_module(name.strip()) if not all(callable(getattr(strategy,k,None)) for k in ('decide','capital_floor')):raise Blocked('HOST_STRATEGY_INTERFACE_MISSING') bars=ROOT/config['bars_path'];macro=ROOT/config['macro_path'];output=ROOT/config['output_path'] year=context['calendar_year'];has_crypto=False if year>=2020 and bars.is_file(): stream=year_bars(bars,year) try:next(stream);has_crypto=True except StopIteration:pass except Blocked as exc: if str(exc)!=f'NO_MARKET_BARS_FOR_YEAR:{year}':raise finally:stream.close() if 1926<=year<=2019 or not has_crypto: with localcontext() as precision: precision.prec=4096 return synthetic_replay(context,strategy,output) return replay(context,strategy,bars,macro,output,require_full_year=False) class ActivityEvents(list): def __init__(self,context): super().__init__() self.context=context from _finbot_supervisor import Store self.store=Store(context['_activity_db']) if context.get('_activity_db') else None def append(self,row): from _finbot_privacy import clean row=clean(row) super().append(row) if self.context.get('_activity_stream'): print(json.dumps({'type':'activity','activity':row}),flush=True) elif self.store: self.store.event('REPLAY_'+row['type'],attempt=self.context['attempt'],year=self.context.get('year'),calendar_year=self.context['calendar_year'],activity=row)