#!/usr/bin/env python3 """ LuckySt Syndicate - Session Summary Builder Fetches trading data from Kalshi API at session end, builds a structured SessionSummary, and optionally generates an AI narrative via Claude Haiku. """ import os import json from dataclasses import dataclass, field, asdict from datetime import datetime, timezone from typing import Optional import httpx # --------------------------------------------------------------------------- # Data structures # --------------------------------------------------------------------------- @dataclass class MarketResult: """Trading result for a single market.""" ticker: str side: str # "yes", "no" position: int total_traded: int # cents avg_cost: Optional[float] = None # cents per contract fills_count: int = 0 volume_contracts: int = 0 @dataclass class SessionSummary: """Complete session summary ready for encryption and on-chain storage.""" instance_id: int script_type: str # "auto1" or "auto2" markets: list[str] = field(default_factory=list) start_time: str = "" end_time: str = "" duration_seconds: int = 0 cycles_completed: int = 0 market_results: list[MarketResult] = field(default_factory=list) estimated_pnl_cents: int = 0 narrative: str = "" def to_json(self) -> str: """Serialize to JSON string (for encryption).""" return json.dumps(asdict(self), indent=2) def to_text(self) -> str: """Human-readable text summary.""" lines = [ f"=== LuckySt Session Report ===", f"Instance: #{self.instance_id} ({self.script_type})", f"Markets: {', '.join(self.markets)}", f"Duration: {self.duration_seconds // 60}m {self.duration_seconds % 60}s", f"Cycles: {self.cycles_completed}", "", ] for mr in self.market_results: avg_str = f" (avg {mr.avg_cost:.0f}c)" if mr.avg_cost else "" lines.append( f" {mr.ticker}: {mr.side.upper()} pos={mr.position} " f"traded={mr.total_traded}c{avg_str} fills={mr.fills_count}" ) lines.append(f"\nEstimated P&L: {self.estimated_pnl_cents}c") if self.narrative: lines.append(f"\n{self.narrative}") lines.append("==============================") return "\n".join(lines) # --------------------------------------------------------------------------- # Kalshi data fetching # --------------------------------------------------------------------------- async def fetch_session_data( kalshi_api, markets: list[str], ) -> dict: """ Fetch positions and fills from Kalshi API for the given markets. Args: kalshi_api: KalshiAPITrader instance (has _request method) markets: List of market tickers Returns: Dict with 'positions' and 'fills' per market """ data = {} for ticker in markets: market_data = {"positions": [], "fills": []} try: # Fetch positions pos_resp = await kalshi_api._request( "GET", f"/portfolio/positions?ticker={ticker}&count_filter=position" ) if isinstance(pos_resp, dict): market_data["positions"] = pos_resp.get("market_positions", []) # Fetch fills fills_resp = await kalshi_api._request( "GET", f"/portfolio/fills?ticker={ticker}" ) if isinstance(fills_resp, dict): market_data["fills"] = fills_resp.get("fills", []) except Exception: pass data[ticker] = market_data return data # --------------------------------------------------------------------------- # AI Narrative # --------------------------------------------------------------------------- async def generate_ai_narrative( data: dict, markets: list[str], duration_seconds: int, cycles: int, ) -> str: """ Generate a 2-3 paragraph narrative summary using Claude Haiku. Falls back to a template if ANTHROPIC_API_KEY is not set. """ api_key = os.environ.get("ANTHROPIC_API_KEY") if not api_key: return _template_narrative(markets, duration_seconds, cycles) prompt = ( f"Write a 2-3 paragraph trading session summary for a market making bot. " f"Markets: {', '.join(markets)}. Duration: {duration_seconds // 60} minutes. " f"Completed {cycles} full market making cycles. " f"Data: {json.dumps(data, default=str)[:2000]}. " f"Be concise and use trading jargon. End with a one-liner about edge." ) try: async with httpx.AsyncClient(timeout=15) as client: resp = await client.post( "https://api.anthropic.com/v1/messages", headers={ "x-api-key": api_key, "anthropic-version": "2023-06-01", "content-type": "application/json", }, json={ "model": "claude-haiku-4-5-20251001", "max_tokens": 300, "messages": [{"role": "user", "content": prompt}], }, ) if resp.status_code == 200: body = resp.json() return body["content"][0]["text"] except Exception: pass return _template_narrative(markets, duration_seconds, cycles) def _template_narrative(markets: list[str], duration_seconds: int, cycles: int) -> str: """Fallback template when AI is unavailable.""" mins = duration_seconds // 60 return ( f"Session ran {mins} minutes across {len(markets)} market(s), " f"completing {cycles} full cycles. " f"Syndicate eternal. Spreads temporary. Edge ours." ) # --------------------------------------------------------------------------- # Summary builder # --------------------------------------------------------------------------- async def build_session_summary( kalshi_api, instance_id: int, markets: list[str], script_type: str, start_time: str, cycles_completed: int, ) -> SessionSummary: """ Build a complete SessionSummary by fetching Kalshi data and generating narrative. Args: kalshi_api: KalshiAPITrader instance instance_id: Trading instance ID markets: List of market tickers script_type: "auto1" or "auto2" start_time: ISO timestamp of session start cycles_completed: Number of completed trading cycles """ end_time = datetime.now(timezone.utc).isoformat() # Calculate duration try: start_dt = datetime.fromisoformat(start_time) end_dt = datetime.fromisoformat(end_time) duration = int((end_dt - start_dt).total_seconds()) except Exception: duration = 0 # Fetch data from Kalshi data = await fetch_session_data(kalshi_api, markets) # Build market results market_results = [] for ticker in markets: mdata = data.get(ticker, {}) positions = mdata.get("positions", []) fills = mdata.get("fills", []) position = 0 total_traded = 0 side = "no" for pos in positions: if pos.get("ticker") == ticker: position = abs(pos.get("position", 0)) total_traded = pos.get("total_traded", 0) side = "yes" if pos.get("position", 0) > 0 else "no" break avg_cost = (total_traded / position) if position > 0 else None market_results.append(MarketResult( ticker=ticker, side=side, position=position, total_traded=total_traded, avg_cost=avg_cost, fills_count=len(fills), volume_contracts=sum(f.get("count", 0) for f in fills), )) # Estimate P&L for 2-market NO strategy estimated_pnl = 0 if len(market_results) == 2 and all(mr.side == "no" for mr in market_results): mr1, mr2 = market_results if mr1.avg_cost and mr2.avg_cost: spread = 100 - mr1.avg_cost - mr2.avg_cost matched = min(mr1.position, mr2.position) estimated_pnl = int(spread * matched) # Generate narrative narrative = await generate_ai_narrative(data, markets, duration, cycles_completed) return SessionSummary( instance_id=instance_id, script_type=script_type, markets=markets, start_time=start_time, end_time=end_time, duration_seconds=duration, cycles_completed=cycles_completed, market_results=market_results, estimated_pnl_cents=estimated_pnl, narrative=narrative, )