#!/usr/bin/env python3 """Read-only FinBot snapshot bridge. Authentication and brokerage remain host-side.""" import argparse from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer import json from pathlib import Path import ssl import urllib.error import urllib.parse import urllib.request ROOT = Path(__file__).resolve().parent DEFAULT_URL = 'https://qzotic-gateway.sotality.com/api/snapshot' ALLOWED_HOSTS = {'qzotic-gateway.sotality.com', 'sotality.com'} MAX_BYTES = 2_000_000 class SnapshotError(ValueError): def __init__(self, code): self.code = code; super().__init__(code) class NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None def validate_url(url): p = urllib.parse.urlsplit(url) if (p.scheme != 'https' or p.hostname not in ALLOWED_HOSTS or p.username or p.password or p.port not in (None, 443) or p.query or p.fragment or p.path not in ('', '/', '/api/snapshot')): raise SnapshotError('INVALID_ENDPOINT') return url def strict_json(raw): def pairs(rows): result = {} for k,v in rows: if k in result: raise SnapshotError('DUPLICATE_FIELD') result[k] = v return result try: return json.loads(raw, object_pairs_hook=pairs, parse_float=Decimal, parse_constant=lambda _: (_ for _ in ()).throw(SnapshotError('INVALID_NUMBER'))) except (ValueError, UnicodeError, RecursionError): raise SnapshotError('INVALID_JSON') from None def stamp(value): if not isinstance(value,str) or len(value)>80: raise SnapshotError('INVALID_TIMESTAMP') try: dt=datetime.fromisoformat(value.replace('Z','+00:00')) except ValueError: raise SnapshotError('INVALID_TIMESTAMP') from None if dt.tzinfo is None: raise SnapshotError('INVALID_TIMESTAMP') return dt.astimezone(timezone.utc).isoformat() def text(value, limit=160): if not isinstance(value,str) or not value.strip() or len(value)>limit or any(ord(c)<32 for c in value): raise SnapshotError('INVALID_TEXT') return value def money(value, signed=False): if type(value) not in (str,int,Decimal) or len(str(value))>80: raise SnapshotError('INVALID_DECIMAL') try: d=Decimal(value) except InvalidOperation: raise SnapshotError('INVALID_DECIMAL') from None if not d.is_finite() or abs(d)>Decimal('1e30') or (not signed and d<0) or d.as_tuple().exponent < -12: raise SnapshotError('INVALID_DECIMAL') return format(d,'f') def normalize(payload, received_at=None): if not isinstance(payload,dict): raise SnapshotError('INVALID_ENVELOPE') if payload.get('status') in ('PACKAGE_RECEIVED_ALERT','PROXY_ACTIVE','VERIFIED_AND_COMMITTED','VERIFICATION_CONFIRMED'): raise SnapshotError('ACKNOWLEDGMENT_ONLY') if payload.get('status') not in (None,'SNAPSHOT_PROCESSED_SUCCESS'): raise SnapshotError('UNSUPPORTED_STATUS') data=payload.get('data',payload) if not isinstance(data,dict): raise SnapshotError('INVALID_ENVELOPE') portfolio=data.get('portfolio') if portfolio is not None and not isinstance(portfolio,dict): raise SnapshotError('INVALID_PORTFOLIO') orders=data.get('orders',data.get('ledger')) research=data.get('research') if research is not None and not isinstance(research,dict): raise SnapshotError('INVALID_RESEARCH') signals=research.get('signals') if research is not None else data.get('signals') if portfolio is None and orders is None and signals is None: raise SnapshotError('NO_SNAPSHOT_DATA') if orders is not None and (not isinstance(orders,list) or len(orders)>1000): raise SnapshotError('INVALID_LEDGER') if signals is not None and (not isinstance(signals,list) or len(signals)>1000): raise SnapshotError('INVALID_SIGNALS') out_portfolio={} for key in ('balance','equity','cash','buying_power'): if portfolio is not None and key in portfolio and portfolio[key] is not None: out_portfolio[key]=money(portfolio[key],signed=True) if portfolio is not None and 'status' in portfolio: out_portfolio['status']=text(portfolio['status']) currency=portfolio.get('currency') if portfolio else None out_portfolio['currency']=text(currency,8) if currency is not None else None mode=data.get('account_mode',payload.get('account_mode','UNKNOWN')) if mode not in ('PAPER','LIVE','UNKNOWN'): raise SnapshotError('INVALID_ACCOUNT_MODE') captured=data.get('captured_at',payload.get('captured_at')) captured=stamp(captured) if captured is not None else None clean_orders=[]; ids=set() for row in orders or []: if not isinstance(row,dict): raise SnapshotError('INVALID_LEDGER_ROW') result={} for key in ('id','symbol','side','status','alpaca_order_id'): v=row.get(key) if v is not None: result[key]=text(str(v) if key=='id' and type(v) is int else v) if 'id' in result: if result['id'] in ids: raise SnapshotError('DUPLICATE_ORDER') ids.add(result['id']) if 'side' in result and result['side'].lower() not in ('buy','sell'): raise SnapshotError('INVALID_SIDE') if 'timestamp' in row: result['timestamp']=stamp(row['timestamp']) if captured and result.get('timestamp') and datetime.fromisoformat(result['timestamp'])>datetime.fromisoformat(captured): raise SnapshotError('ORDER_AFTER_SNAPSHOT') for key in ('qty','price','allocation_spent'): if row.get(key) is not None: result[key]=money(row[key]) if not result: raise SnapshotError('UNSUPPORTED_LEDGER_ROW') clean_orders.append(result) clean_signals=[] for row in signals or []: if isinstance(row,str): clean_signals.append({'signal':text(row,300)}); continue if not isinstance(row,dict): raise SnapshotError('INVALID_SIGNAL') result={} for key in ('symbol','signal','action','status','reason','strategy'): if row.get(key) is not None: result[key]=text(row[key],300) for key in ('price','confidence','score'): if row.get(key) is not None: result[key]=money(row[key],signed=key=='score') if row.get('timestamp') is not None: result['timestamp']=stamp(row['timestamp']) if not result: raise SnapshotError('UNSUPPORTED_SIGNAL_ROW') clean_signals.append(result) return {'schema_version':2,'received_at':received_at or datetime.now(timezone.utc).isoformat(), 'captured_at':captured,'account_mode':mode,'portfolio':out_portfolio, 'orders':clean_orders,'signals':clean_signals, 'availability':{'portfolio':portfolio is not None,'orders':orders is not None,'signals':signals is not None}, 'assurance':'HOST_REPORTED','capabilities':{'submit_orders':False,'liquidate':False}} def fetch_snapshot(url): validate_url(url) opener=urllib.request.build_opener(NoRedirect(),urllib.request.HTTPSHandler(context=ssl.create_default_context())) try: request=urllib.request.Request(url,headers={'Accept':'application/json','Cache-Control':'no-cache'},method='GET') with opener.open(request,timeout=10) as response: if response.status!=200: raise SnapshotError('UPSTREAM_HTTP_ERROR') if response.headers.get('Content-Type','').split(';')[0].strip().lower()!='application/json': raise SnapshotError('EXPECTED_JSON') raw=response.read(MAX_BYTES+1) if len(raw)>MAX_BYTES: raise SnapshotError('RESPONSE_TOO_LARGE') return normalize(strict_json(raw)) except urllib.error.HTTPError as error: code=error.code;error.close() raise SnapshotError({401:'AUTHORIZATION_REQUIRED',403:'ACCESS_DENIED',429:'RATE_LIMITED'}.get(code,'REDIRECT_REJECTED' if 300<=code<400 else 'UPSTREAM_HTTP_ERROR')) from None except (urllib.error.URLError,OSError,TimeoutError): raise SnapshotError('CONNECTION_UNAVAILABLE') from None class Handler(BaseHTTPRequestHandler): def log_message(self,*args): pass def setup(self): super().setup();self.connection.settimeout(12) def reply(self,status,body,mime='application/json'): raw=json.dumps(body).encode() if mime=='application/json' else body self.send_response(status);self.send_header('Content-Type',mime);self.send_header('Content-Length',str(len(raw))) self.send_header('Cache-Control','no-store');self.send_header('X-Content-Type-Options','nosniff') self.send_header('Content-Security-Policy',"default-src 'self'; connect-src 'self'; script-src 'self'; style-src 'self'; object-src 'none'; frame-ancestors 'none'") self.send_header('Connection','close');self.end_headers();self.wfile.write(raw);self.close_connection=True def do_GET(self): host=f'127.0.0.1:{self.server.server_address[1]}' if self.headers.get('Host')!=host or self.headers.get('Origin') not in (None,'http://'+host): self.reply(403,{'error':'PROXY_ACTIVE'});return if self.path=='/api/snapshot': if not self.server.fetch_lock.acquire(blocking=False): self.reply(429,{'error':'REQUEST_IN_PROGRESS'});return try: try:self.reply(200,{'snapshot':self.server.fetcher(self.server.endpoint)}) except SnapshotError as error:self.reply(502,{'error':error.code}) except Exception:self.reply(502,{'error':'SNAPSHOT_UNAVAILABLE'}) finally:self.server.fetch_lock.release() return assets={'/':('dashboard.html','text/html; charset=utf-8'),'/dashboard.js':('dashboard.js','text/javascript; charset=utf-8'),'/dashboard.css':('dashboard.css','text/css; charset=utf-8')} if self.path not in assets:self.reply(404,{'error':'NOT_FOUND'});return name,mime=assets[self.path];self.reply(200,(ROOT/name).read_bytes(),mime) def do_POST(self):self.reply(405,{'error':'READ_ONLY_CLIENT'}) def create_server(endpoint=DEFAULT_URL,port=8870,fetcher=fetch_snapshot): import threading validate_url(endpoint) server=ThreadingHTTPServer(('127.0.0.1',port),Handler) server.endpoint=endpoint;server.fetcher=fetcher;server.fetch_lock=threading.Lock() return server def main(): parser=argparse.ArgumentParser(description=__doc__) parser.add_argument('--snapshot-url',default=DEFAULT_URL) parser.add_argument('--port',type=int,default=8870) parser.add_argument('--open-browser',action='store_true') parser.add_argument('--check',action='store_true',help='One read-only request; print status and counts only') args=parser.parse_args() if args.check: try: data=fetch_snapshot(args.snapshot_url) print(json.dumps({'result':'SNAPSHOT_PARSED','account_mode':data['account_mode'],'source_timestamp_present':data['captured_at'] is not None,'order_count':len(data['orders']),'signal_count':len(data['signals']),'orders_submitted':0})) except SnapshotError as error: print(json.dumps({'result':error.code,'orders_submitted':0}));raise SystemExit(1) return with create_server(args.snapshot_url,args.port) as server: print(f'Open http://127.0.0.1:{server.server_address[1]}/ — click Connect to request snapshots.',flush=True) if args.open_browser: import webbrowser webbrowser.open(f'http://127.0.0.1:{server.server_address[1]}/') try:server.serve_forever() except KeyboardInterrupt:pass if __name__=='__main__': try:main() except (OSError,ValueError):raise SystemExit('Client not started: check endpoint and available local port.')