Upload builderbrain/reasoning_agent.py

builderbrain/reasoning_agent.py

@@ -0,0 +1,401 @@
+"""
+Reasoning Agent + Trade Signal Generation
+=========================================
+
+Produces structured reasoning traces for each trade recommendation.
+Each trace is:
+- Hashed and anchored on-chain as an artifact
+- Contains: data sources, argument structure, risk factors, confidence
+- Links to builder code execution for auditability
+
+This is "Trading-R1" — reasoning as a first-class product.
+"""
+
+import hashlib
+import json
+import time
+from dataclasses import dataclass, asdict
+from typing import List, Dict, Optional, Any
+from datetime import datetime
+import uuid
+
+
+@dataclass
+class DataSource:
+    """A piece of evidence used in reasoning."""
+    source_type: str  # 'polymarket', 'news', 'social', 'onchain', 'model'
+    source_id: str  # URL, API endpoint, tweet ID, etc.
+    timestamp: str
+    data_summary: str
+    relevance_score: float  # 0-1
+    raw_data: Optional[Dict] = None
+
+
+@dataclass
+class Argument:
+    """A structured argument for a position."""
+    claim: str
+    evidence: List[str]  # source_ids
+    strength: float  # 0-1
+    direction: str  # 'bullish', 'bearish', 'neutral'
+    confidence: float  # 0-1
+
+
+@dataclass
+class RiskFactor:
+    """A risk that could invalidate the thesis."""
+    description: str
+    probability: float  # 0-1
+    impact: str  # 'low', 'medium', 'high', 'catastrophic'
+    mitigation: str
+
+
+@dataclass
+class ReasoningTrace:
+    """
+    A complete reasoning artifact for a trade recommendation.
+    
+    Anchored on-chain via hash for auditability.
+    """
+    trace_id: str
+    market_id: str
+    market_title: str
+    side: str  # 'YES' or 'NO'
+    timestamp: str
+    
+    # Core reasoning
+    model_probability: float
+    market_probability: float
+    edge: float
+    
+    # Components
+    data_sources: List[DataSource]
+    arguments: List[Argument]
+    risk_factors: List[RiskFactor]
+    
+    # Meta
+    agent_version: str
+    confidence: float  # composite 0-1
+    reasoning_hash: str  # SHA256 of canonical JSON
+    
+    # Execution link
+    builder_code: Optional[str] = None
+    executed: bool = False
+    execution_tx: Optional[str] = None
+
+
+@dataclass
+class TradeSignal:
+    """A complete trade recommendation with reasoning."""
+    market_id: str
+    side: str
+    size_fraction: float  # of bankroll
+    expected_return: float
+    confidence: float
+    reasoning_trace: ReasoningTrace
+    urgency: str  # 'immediate', '24h', 'week', 'pass'
+    
+    def to_dict(self) -> Dict:
+        return asdict(self)
+
+
+class ReasoningAgent:
+    """
+    Generates structured reasoning traces for prediction market trades.
+    
+    Simulates the intelligence layer: ingesting data, forming beliefs,
+    articulating arguments, and quantifying risks.
+    
+    In production, this would connect to live data feeds (news APIs,
+    social media, on-chain signals). For hackathon, we simulate with
+    structured inputs.
+    """
+    
+    def __init__(self, agent_version: str = "builderbrain-v0.1"):
+        self.agent_version = agent_version
+        self.trace_history: List[ReasoningTrace] = []
+        self.knowledge_base: Dict[str, Any] = {}
+    
+    # ────────────────────────────── Core Reasoning ──────────────────────────────
+    
+    def reason_about_market(
+        self,
+        market_id: str,
+        market_title: str,
+        market_prob: float,
+        model_prob: float,
+        data_sources: List[Dict],
+        theme: str = "general",
+    ) -> ReasoningTrace:
+        """
+        Generate a complete reasoning trace for a market.
+        
+        In production, this would:
+        1. Scrape news/social for relevant signals
+        2. Run NLP models for sentiment/entity extraction
+        3. Cross-reference with historical market patterns
+        4. Produce probability estimate with uncertainty
+        
+        For hackathon, we simulate with structured inputs.
+        """
+        edge = model_prob - market_prob
+        side = "YES" if edge > 0 else "NO"
+        
+        # Parse data sources
+        sources = [DataSource(**ds) for ds in data_sources]
+        
+        # Generate arguments based on edge direction and theme
+        arguments = self._generate_arguments(
+            market_title, theme, edge, sources
+        )
+        
+        # Generate risk factors
+        risks = self._generate_risks(market_title, theme, edge)
+        
+        # Compute composite confidence
+        arg_confidence = max(
+            [a.confidence for a in arguments] + [0.5]
+        )
+        data_quality = min(1.0, len(sources) * 0.2 + 0.3)
+        confidence = arg_confidence * data_quality * min(abs(edge) * 5, 1.0)
+        
+        # Build trace
+        trace = ReasoningTrace(
+            trace_id=f"trace_{uuid.uuid4().hex[:12]}",
+            market_id=market_id,
+            market_title=market_title,
+            side=side,
+            timestamp=datetime.utcnow().isoformat(),
+            model_probability=model_prob,
+            market_probability=market_prob,
+            edge=edge,
+            data_sources=sources,
+            arguments=arguments,
+            risk_factors=risks,
+            agent_version=self.agent_version,
+            confidence=round(confidence, 4),
+            reasoning_hash="",  # computed below
+        )
+        
+        # Compute hash
+        trace.reasoning_hash = self._hash_trace(trace)
+        
+        self.trace_history.append(trace)
+        return trace
+    
+    def _generate_arguments(
+        self,
+        title: str,
+        theme: str,
+        edge: float,
+        sources: List[DataSource],
+    ) -> List[Argument]:
+        """Generate structured arguments from market context."""
+        arguments = []
+        
+        # Base argument from edge direction
+        if edge > 0:
+            arguments.append(Argument(
+                claim=f"Market underprices {title} by {abs(edge):.1%}",
+                evidence=[s.source_id for s in sources[:2]],
+                strength=min(abs(edge) * 3, 0.95),
+                direction="bullish" if edge > 0 else "bearish",
+                confidence=min(abs(edge) * 2, 0.9),
+            ))
+        
+        # Theme-specific arguments
+        theme_args = self._theme_arguments(title, theme, edge, sources)
+        arguments.extend(theme_args)
+        
+        return arguments
+    
+    def _theme_arguments(
+        self,
+        title: str,
+        theme: str,
+        edge: float,
+        sources: List[DataSource],
+    ) -> List[Argument]:
+        """Generate theme-specific arguments."""
+        args = []
+        
+        if theme == "politics":
+            args.append(Argument(
+                claim="Polling momentum and fundraising data support this direction",
+                evidence=[s.source_id for s in sources if s.source_type == "news"][:2],
+                strength=0.7,
+                direction="bullish" if edge > 0 else "bearish",
+                confidence=0.65,
+            ))
+        
+        elif theme == "crypto":
+            args.append(Argument(
+                claim="On-chain flows and ETF momentum align with price direction",
+                evidence=[s.source_id for s in sources if s.source_type == "onchain"][:2],
+                strength=0.75,
+                direction="bullish" if edge > 0 else "bearish",
+                confidence=0.7,
+            ))
+        
+        elif theme == "sports":
+            args.append(Argument(
+                claim="Injury reports and lineup data support this probability",
+                evidence=[s.source_id for s in sources if s.source_type == "news"][:2],
+                strength=0.6,
+                direction="bullish" if edge > 0 else "bearish",
+                confidence=0.55,
+            ))
+        
+        elif theme == "macro":
+            args.append(Argument(
+                claim="Fed communications and economic prints support this direction",
+                evidence=[s.source_id for s in sources if s.source_type == "news"][:2],
+                strength=0.65,
+                direction="bullish" if edge > 0 else "bearish",
+                confidence=0.6,
+            ))
+        
+        return args
+    
+    def _generate_risks(
+        self,
+        title: str,
+        theme: str,
+        edge: float,
+    ) -> List[RiskFactor]:
+        """Generate risk factors for a market."""
+        risks = [
+            RiskFactor(
+                description="Black swan event invalidates base case",
+                probability=0.05,
+                impact="catastrophic",
+                mitigation="Position sizing limits + correlation caps",
+            ),
+            RiskFactor(
+                description="New information shifts probability before position closes",
+                probability=0.25,
+                impact="medium",
+                mitigation="Dynamic position updates + stop-loss on edge decay",
+            ),
+            RiskFactor(
+                description="Market manipulation or wash trading distorts price",
+                probability=0.1,
+                impact="high",
+                mitigation="Liquidity filters + cross-market validation",
+            ),
+        ]
+        
+        if theme == "politics":
+            risks.append(RiskFactor(
+                description="Late-breaking scandal or debate performance shift",
+                probability=0.2,
+                impact="high",
+                mitigation="Reduce position 48h before major events",
+            ))
+        
+        elif theme == "crypto":
+            risks.append(RiskFactor(
+                description="Regulatory action (SEC, exchange shutdown)",
+                probability=0.15,
+                impact="catastrophic",
+                mitigation="Diversify across uncorrelated tokens + max 10% per token",
+            ))
+        
+        return risks
+    
+    def _hash_trace(self, trace: ReasoningTrace) -> str:
+        """Compute SHA256 hash of canonical trace representation."""
+        # Create canonical JSON (sorted keys, no whitespace)
+        canonical = json.dumps({
+            "market_id": trace.market_id,
+            "side": trace.side,
+            "model_prob": trace.model_probability,
+            "market_prob": trace.market_probability,
+            "edge": trace.edge,
+            "arguments": [
+                {"claim": a.claim, "strength": a.strength, "confidence": a.confidence}
+                for a in trace.arguments
+            ],
+            "risks": [
+                {"desc": r.description, "prob": r.probability, "impact": r.impact}
+                for r in trace.risk_factors
+            ],
+            "timestamp": trace.timestamp,
+        }, sort_keys=True, separators=(',', ':'))
+        
+        return hashlib.sha256(canonical.encode()).hexdigest()[:32]
+    
+    # ────────────────────────────── Signal Generation ──────────────────────────────
+    
+    def generate_signal(
+        self,
+        trace: ReasoningTrace,
+        kelly_fraction: float,
+        expected_return: float,
+    ) -> TradeSignal:
+        """Convert reasoning trace to executable trade signal."""
+        # Determine urgency based on edge magnitude and time to expiry
+        abs_edge = abs(trace.edge)
+        if abs_edge > 0.15:
+            urgency = "immediate"
+        elif abs_edge > 0.08:
+            urgency = "24h"
+        elif abs_edge > 0.03:
+            urgency = "week"
+        else:
+            urgency = "pass"
+        
+        return TradeSignal(
+            market_id=trace.market_id,
+            side=trace.side,
+            size_fraction=kelly_fraction,
+            expected_return=expected_return,
+            confidence=trace.confidence,
+            reasoning_trace=trace,
+            urgency=urgency,
+        )
+    
+    # ────────────────────────────── Trace Retrieval ──────────────────────────────
+    
+    def get_trace(self, trace_id: str) -> Optional[ReasoningTrace]:
+        """Retrieve a trace by ID."""
+        for t in self.trace_history:
+            if t.trace_id == trace_id:
+                return t
+        return None
+    
+    def get_traces_for_market(self, market_id: str) -> List[ReasoningTrace]:
+        """Get all traces for a market."""
+        return [t for t in self.trace_history if t.market_id == market_id]
+    
+    def get_top_traces(
+        self,
+        min_confidence: float = 0.6,
+        limit: int = 20,
+    ) -> List[ReasoningTrace]:
+        """Get highest-confidence traces."""
+        filtered = [t for t in self.trace_history if t.confidence >= min_confidence]
+        filtered.sort(key=lambda t: t.confidence, reverse=True)
+        return filtered[:limit]
+    
+    def export_traces(self, filepath: str):
+        """Export all traces to JSON for audit."""
+        data = [asdict(t) for t in self.trace_history]
+        with open(filepath, 'w') as f:
+            json.dump(data, f, indent=2, default=str)
+    
+    def stats(self) -> Dict:
+        """Agent performance statistics."""
+        if not self.trace_history:
+            return {}
+        
+        edges = [t.edge for t in self.trace_history]
+        confidences = [t.confidence for t in self.trace_history]
+        
+        return {
+            "total_traces": len(self.trace_history),
+            "avg_edge": sum(edges) / len(edges),
+            "avg_confidence": sum(confidences) / len(confidences),
+            "high_confidence_traces": sum(1 for c in confidences if c > 0.7),
+            "themes": list(set(t.market_id.split('_')[0] for t in self.trace_history)),
+        }