"""Independent account-2 flow auditor. No exchange write capability.""" import hashlib import hmac import importlib import json import os from pathlib import Path import sqlite3 import subprocess import sys import time from datetime import datetime, timezone, timedelta from urllib.parse import urlencode import requests ROOT = Path('/opt/charter-live/live_deploy_v2') HOME = ROOT / 'flow_certifier' OUTPUT = ROOT / 'logs/flow_certification.json' def utc(): return datetime.now(timezone.utc) def parse(value): return datetime.fromisoformat(str(value).replace('Z', '+00:00')).astimezone(timezone.utc) def evaluate(snapshot, now): faults = list(snapshot.get('faults', [])) heartbeat = snapshot.get('controller', {}) try: age = (now-parse(heartbeat['generated_at'])).total_seconds() if not -30 <= age <= 900: faults.append({'code':'CONTROLLER_HEARTBEAT_STALE', 'age_seconds':round(age)}) except (KeyError, ValueError, TypeError): faults.append({'code':'CONTROLLER_HEARTBEAT_MISSING'}) known = set(snapshot.get('known_order_ids', [])) manual = set(snapshot.get('manual_order_ids', [])) acknowledged = set(snapshot.get('operator_acknowledged_order_ids', [])) # Solo gli ordini creati dall'operatore/API possono essere collegati a un alert. # CreateByClosing / CreateByTakeProfit / CreateByStopLoss / CreateByStopOrder sono # generati dall'exchange e NON hanno e non possono avere un webhook. unmatched = [o for o in snapshot.get('orders', []) if o.get('createType') == 'CreateByUser' and o['orderId'] not in known | manual | acknowledged] declared = set(heartbeat.get('unmatched_order_ids', [])) for order in unmatched: faults.append({'code':'ORDER_WITHOUT_WEBHOOK', 'order_id':order['orderId'], 'symbol':order['symbol']}) # Allow a full controller cycle before calling this a detection miss. if int(order.get('createdTime') or 0)/1000 < now.timestamp()-900 and order['orderId'] not in declared: faults.append({'code':'CONTROLLER_DETECTION_MISS', 'order_id':order['orderId']}) owners = {x['symbol']:x for x in snapshot.get('owners', [])} for position in snapshot.get('positions', []): owner = owners.get(position['symbol']) side = {'Buy':'long','Sell':'short'}.get(position['side']) if not owner or owner.get('side') != side: faults.append({'code':'POSITION_SIDE_MISMATCH','symbol':position['symbol'], 'expected':owner.get('side') if owner else None,'actual':side}) return {'agent':'Flow Certifier Account 2','generated_at':now.isoformat(), 'valid_until':(now+timedelta(seconds=180)).isoformat(), 'status':'NOT_CERTIFIED' if faults else 'CERTIFIED', 'read_only':True,'faults':faults,'checks':snapshot.get('checks',{}), 'controller_generated_at':heartbeat.get('generated_at'), 'coverage':'All account-2 assets; received webhooks, live positions and executed Bybit orders in last 24h. TradingView pre-ingress state not certified.', 'manual_order_ids':sorted(manual), 'operator_acknowledged_order_ids':sorted(acknowledged)} class ReadOnlyBybit: def __init__(self): from dotenv import dotenv_values env=dotenv_values(ROOT/'.env.vps') self.key=env['BYBIT_HETZNER_API_KEY'];self.secret=env['BYBIT_HETZNER_SECRET'] self.session=requests.Session() def get(self,path,params): if path not in ('/v5/order/history','/v5/position/list'): raise RuntimeError('Read-only endpoint not allowed') query=urlencode(params);stamp=str(int(time.time()*1000)) signature=hmac.new(self.secret.encode(),(stamp+self.key+'5000'+query).encode(),hashlib.sha256).hexdigest() response=self.session.get('https://api-demo.bybit.com'+path+'?'+query, headers={'X-BAPI-API-KEY':self.key,'X-BAPI-SIGN':signature,'X-BAPI-TIMESTAMP':stamp,'X-BAPI-RECV-WINDOW':'5000'},timeout=10) response.raise_for_status();data=response.json() if data.get('retCode')!=0:raise RuntimeError('Bybit read failure code='+str(data.get('retCode'))) return data['result'] def pages(self,path,params): rows=[];seen=set() for _ in range(20): data=self.get(path,params);rows.extend(data['list']);cursor=data.get('nextPageCursor') if not cursor:return rows if cursor in seen:raise RuntimeError('Repeated pagination cursor') seen.add(cursor);params=dict(params,cursor=cursor) raise RuntimeError('Incomplete pagination') def guard_probe(): sys.path.insert(0,str(ROOT)) client_module=importlib.import_module('bybit_demo_client') class Trap: calls=0 def post(self,*a,**kw): self.calls+=1 raise RuntimeError('PROBE_TRANSPORT_ATTEMPT') client=client_module.BybitDemoClient.__new__(client_module.BybitDemoClient) client.api_key='probe';client.secret='probe';client.session=Trap() def attempt(reduce): try: client._request('POST','/v5/order/create',{'category':'linear','symbol':'PROBEUSDT','side':'Sell','qty':'0','reduceOnly':reduce},signed=True) return None except RuntimeError as exc: return str(exc) # reduceOnly=False PUO' aprire una posizione: DEVE essere bloccato dal guard. err=attempt(False) if not (err and err.startswith('ORDER_DISPATCH_BLOCKED')): raise RuntimeError('Unlinked non-reduce order not blocked: %r' % err) # reduceOnly=True e' esente dal link: DEVE arrivare al transport (che nella sonda e' finto). err=attempt(True) if not (err and err.startswith('PROBE_TRANSPORT_ATTEMPT')): raise RuntimeError('Exempt reduceOnly order did not reach transport: %r' % err) return True def snapshot(): result={'faults':[], 'checks':{}} policy=json.loads((HOME/'policy.json').read_text()) result['manual_order_ids']=list(policy['manual_order_ids']) # Read the operator's current dashboard decisions at every cycle. Exact IDs only. dashboard_policy=json.loads((ROOT/'v2_stats_trade_policy.json').read_text()) mapping=dashboard_policy.get('strategy_by_closed_order_id', {}) sources=dashboard_policy.get('classification_source_by_closed_order_id', {}) certified={oid for oid in mapping if sources.get(oid)=='operator_certified_from_dashboard' or str(sources.get(oid, '')).startswith('review_proposal:')} result['manual_order_ids'].extend(oid for oid in certified if mapping[oid]=='manuale') acknowledged=set(certified) queue=json.loads((ROOT/'v2_classification_review_queue.json').read_text()) latest={item['order_id']:item for item in queue if item.get('order_id') and item.get('action') in ('classify','exclude_test','exclude_operational','needs_review')} acknowledged.update(oid for oid,item in latest.items() if item.get('status') in ('applied','certified','resolved') and item.get('action') in ('classify','exclude_test','exclude_operational')) result['operator_acknowledged_order_ids']=sorted(acknowledged) result['checks']['dashboard_decisions_read_ok']=True for name,digest in policy['source_sha256'].items(): path=ROOT/name if not path.is_file() or hashlib.sha256(path.read_bytes()).hexdigest()!=digest: result['faults'].append({'code':'GUARD_SOURCE_CHANGED','file':name}) result['checks']['source_hashes_ok']=not result['faults'] try: result['checks']['unlinked_orders_blocked']=guard_probe() except Exception as exc: result['faults'].append({'code':'GUARD_PROBE_FAILED','error_type':type(exc).__name__}) state=subprocess.check_output(['systemctl','show','charter-v2-webhook','-p','ActiveState','-p','MainPID','-p','UnitFileState','-p','Restart'],text=True) props=dict(line.split('=',1) for line in state.splitlines() if '=' in line) result['checks']['webhook_service']=props if props.get('ActiveState')!='active' or props.get('UnitFileState')!='enabled': result['faults'].append({'code':'WEBHOOK_SERVICE_NOT_PERSISTENT'}) if props.get('Restart') not in ('always','on-failure'): result['faults'].append({'code':'WEBHOOK_RESTART_POLICY_MISSING'}) try: stat=Path('/proc/'+str(int(props['MainPID']))+'/stat').read_text() ticks=float(stat.split(') ',1)[1].split()[19]) boot=float(next(line.split()[1] for line in Path('/proc/stat').read_text().splitlines() if line.startswith('btime '))) started=boot+ticks/os.sysconf('SC_CLK_TCK') if any((ROOT/name).stat().st_mtime>started+2 for name in policy['source_sha256']): result['faults'].append({'code':'WEBHOOK_RUNNING_OLD_CODE'}) response=requests.get('http://127.0.0.1:5581/health',timeout=3) response.raise_for_status() if not response.json().get('ok'):raise RuntimeError('Unhealthy webhook') except Exception as exc: result['faults'].append({'code':'WEBHOOK_RUNTIME_CHECK_FAILED','error_type':type(exc).__name__}) path=ROOT/'logs/controller_attestation.json' if path.exists(): try:result['controller']=json.loads(path.read_text()) except (ValueError,OSError):pass db=sqlite3.connect('file:'+str(ROOT/'logs/webhook_queue.db')+'?mode=ro',uri=True,timeout=5) db.row_factory=sqlite3.Row try: result['known_order_ids']=[row[0] for row in db.execute('SELECT order_id FROM orders UNION SELECT order_id FROM queue WHERE order_id IS NOT NULL')] result['owners']=[dict(row) for row in db.execute('SELECT symbol,side,strategy,entry_order_id FROM position_ownership')] for row in db.execute("SELECT request_id,received_at FROM queue WHERE status='pending'"): if (utc()-parse(row['received_at'])).total_seconds()>600: result['faults'].append({'code':'STALE_PENDING_REQUEST','request_id':row['request_id']}) finally:db.close() client=ReadOnlyBybit() result['positions']=[p for p in client.pages('/v5/position/list',{'category':'linear','settleCoin':'USDT','limit':'200'}) if float(p.get('size') or 0)>0] result['orders']=[o for o in client.pages('/v5/order/history',{'category':'linear','startTime':str(int((utc()-timedelta(hours=24)).timestamp()*1000)),'limit':'100'}) if float(o.get('cumExecQty') or 0)>0] result['checks']['bybit_read_ok']=True return result def main(): try:report=evaluate(snapshot(),utc()) except Exception as exc: report=evaluate({'faults':[{'code':'CERTIFIER_CHECK_FAILED','error_type':type(exc).__name__}]},utc()) OUTPUT.parent.mkdir(exist_ok=True) temp=OUTPUT.with_suffix('.tmp');temp.write_text(json.dumps(report,indent=2),encoding='utf-8');os.replace(temp,OUTPUT) print(json.dumps({'status':report['status'],'generated_at':report['generated_at'],'fault_codes':sorted({f['code'] for f in report['faults']})})) if __name__=='__main__':main()