#!/usr/bin/env python3 import json, os, math, subprocess from pathlib import Path from datetime import datetime, timezone from decimal import Decimal, InvalidOperation from collections import Counter, defaultdict ROOT=Path('/home/agent/projects/CryptoTradingBot/Crypto_Agent') START=datetime.fromisoformat('2026-07-04T23:10:00+02:00').astimezone(timezone.utc) GATE_SPLIT=datetime.fromisoformat('2026-07-05T07:01:18+02:00').astimezone(timezone.utc) strategies=['candidate_v76_strict_live_candidate','candidate_v76_research_probe','candidate_v76_fee_aware_anti_chase','candidate_swing_trend_retest_research'] def dt(s): if not s: return None try: return datetime.fromisoformat(str(s).replace('Z','+00:00')).astimezone(timezone.utc) except Exception: return None def dec(x, default=Decimal('0')): try: return Decimal(str(x)) except Exception: return default def read_jsonl(p): if not p.exists(): return with p.open() as f: for line in f: line=line.strip() if not line: continue try: yield json.loads(line) except Exception: continue def summarize_strategy(sid): base=ROOT/'runtime/experiments'/sid trades=[r for r in read_jsonl(base/'trade_journal.jsonl') or [] if (dt(r.get('timestamp')) and dt(r.get('timestamp'))>=START)] signals=[r for r in read_jsonl(base/'signal_journal.jsonl') or [] if (dt(r.get('timestamp')) and dt(r.get('timestamp'))>=START)] # all entries indexed approximate for R by coin FIFO all_rows=list(read_jsonl(base/'trade_journal.jsonl') or []) fifo=defaultdict(list) closed=[] for r in all_rows: t=dt(r.get('timestamp')) ev=r.get('event') coin=r.get('coin') if ev=='entry': fifo[coin].append(r) elif ev=='exit': ent=fifo[coin].pop(0) if fifo[coin] else {} if t and t>=START: pnl=dec(r.get('realized_pnl_usd', r.get('net_pnl_usd',0))) entry=dec(ent.get('entry_price', r.get('entry_price',0))) stop=dec(ent.get('stop_loss', r.get('stop_loss',0))) size=dec(r.get('size', ent.get('size',0))) risk=abs(entry-stop)*size rr=(pnl/risk) if risk else Decimal('0') closed.append((r,ent,pnl,rr)) wins=[x for x in closed if x[2]>0]; losses=[x for x in closed if x[2]<0] gross_win=sum((x[2] for x in wins), Decimal('0')); gross_loss=sum((-x[2] for x in losses), Decimal('0')) state={} try: state=json.loads((base/'state.json').read_text()) except Exception: pass prev={k:dec(v) for k,v in state.get('prev_mids',{}).items()} openpos=state.get('open_positions') or state.get('positions') or {} unreal=Decimal('0'); open_details=[] if isinstance(openpos, dict): for coin,pos in openpos.items(): if not isinstance(pos, dict): continue entry=dec(pos.get('entry_price', pos.get('entry',0))); size=dec(pos.get('size', pos.get('qty',0))) side=str(pos.get('side','long')).lower(); mark=prev.get(coin, dec(pos.get('mark',entry))) u=(mark-entry)*size*(Decimal('-1') if side=='short' else Decimal('1')) unreal+=u; open_details.append({'coin':coin,'side':side,'entry':str(entry),'mark':str(mark),'unreal':str(u)}) block_counts=Counter(); decision_counts=Counter(); coins=Counter(); mtrend=Counter(); recent=Counter() for r in signals: coins[r.get('coin','?')]+=1 decision_counts[r.get('final_decision') or r.get('decision') or ('would_enter' if r.get('would_enter') else 'blocked')]+=1 br=r.get('block_reason') or r.get('block_reasons') or [] if isinstance(br,str): br=[br] for b in br: block_counts[str(b)]+=1 rg=r.get('risk_gate_result') or {} for b in (rg.get('reasons') or []): block_counts[str(b)]+=1 if r.get('trend_state'): mtrend[str(r.get('trend_state'))]+=1 for key in ['recent_stop_cooldown','risk_reward_below_1_4x','market_trend_down','market_breadth_not_supportive']: if key in str(br) or key in str(rg.get('reasons') or []): recent[key]+=1 reasons=Counter([x[0].get('exit_reason','?') for x in closed]) closed_coins=Counter([x[0].get('coin','?') for x in closed]) return dict(sid=sid, closed=len(closed), wins=len(wins), losses=len(losses), pnl=str(sum((x[2] for x in closed),Decimal('0'))), r=str(sum((x[3] for x in closed),Decimal('0'))), winrate=(len(wins)/len(closed)*100 if closed else 0), pf=(float(gross_win/gross_loss) if gross_loss else (float('inf') if gross_win else 0)), avg_win=str(gross_win/len(wins) if wins else Decimal('0')), avg_loss=str(-gross_loss/len(losses) if losses else Decimal('0')), reasons=reasons, closed_coins=closed_coins, open_count=len(open_details), unreal=str(unreal), open_details=open_details, blocks=block_counts, keyblocks=recent, decisions=decision_counts, trends=mtrend, trade_events=len(trades), signal_rows=len(signals), paper_flags={str(k):v for k,v in Counter((r.get('paper_trading'),r.get('research'),r.get('mainnet_signed_action')) for r in trades[:50]).items()}) def summarize_desk(): base=ROOT/'runtime/experiments/trader_desk_shadow' decs=[r for r in read_jsonl(base/'decision_journal.jsonl') or [] if dt(r.get('timestamp')) and dt(r.get('timestamp'))>=START] outs=[r for r in read_jsonl(base/'outcome_journal.jsonl') or [] if dt(r.get('timestamp')) and dt(r.get('timestamp'))>=START] closes=[r for r in outs if r.get('event')=='shadow_close'] opens=[r for r in outs if r.get('event')=='shadow_open'] rvals=[dec(r.get('r_multiple')) for r in closes] wins=[x for x in rvals if x>0]; losses=[x for x in rvals if x<0] block=Counter(); actions=Counter(); directions=Counter(); coins=Counter(); reasons=Counter(); buckets=Counter() for r in decs: actions[r.get('action','?')]+=1; directions[r.get('direction','?')]+=1; coins[r.get('coin','?')]+=1 bs=r.get('blockers') or [] if isinstance(bs,str): bs=[bs] for b in bs: block[str(b)]+=1 for r in closes: reasons[r.get('reason','?')]+=1; buckets[r.get('bucket','?')]+=1 state={} try: state=json.loads((base/'state.json').read_text()) except Exception: pass return dict(decisions=len(decs), opens=len(opens), closed=len(closes), wins=len(wins), losses=len(losses), scratch=sum(1 for x in rvals if x==0), r=str(sum(rvals,Decimal('0'))), winrate=(len(wins)/len(closes)*100 if closes else 0), pf=(float(sum(wins,Decimal('0'))/(-sum(losses,Decimal('0')))) if losses else (float('inf') if wins else 0)), avg_win=str(sum(wins,Decimal('0'))/len(wins) if wins else Decimal('0')), avg_loss=str(sum(losses,Decimal('0'))/len(losses) if losses else Decimal('0')), short_share=(directions.get('short',0)/len(decs)*100 if decs else 0), directions=directions, coins=coins, blockers=block, reasons=reasons, buckets=buckets, state=state) def anti_chase_split(): base=ROOT/'runtime/experiments/candidate_v76_fee_aware_anti_chase' rows=list(read_jsonl(base/'trade_journal.jsonl') or []) out={} for name,a,b in [('vor_neue_gates',START,GATE_SPLIT),('nach_neue_gates',GATE_SPLIT,datetime.max.replace(tzinfo=timezone.utc))]: exits=[r for r in rows if r.get('event')=='exit' and (t:=dt(r.get('timestamp'))) and a<=t