Add root app.py for Streamlit GUI and dependencies

app.py

@@ -0,0 +1,67 @@
+import streamlit as st
+import asyncio
+import pandas as pd
+from datetime import datetime, timezone
+
+from src.strategies.arbitrage import CrossPlatformArbitrage
+# We would import the live clients here, but for the dashboard
+# we can instantiate the scanner and populate it directly for testing.
+
+st.set_page_config(page_title="ArbIntel Scanner", layout="wide", page_icon="📈")
+
+st.title("ArbIntel: Prediction Markets Alpha Engine")
+st.markdown("### Live Cross-Platform Arbitrage Scanner")
+st.write("Detecting price inefficiencies between Polymarket and Kalshi in real-time.")
+
+async def fetch_and_scan():
+    """Simulate fetching live data and running the scanner"""
+    arb_engine = CrossPlatformArbitrage(min_profit_threshold=0.005) # 0.5% threshold
+    
+    # In a real scenario, this would await client responses.
+    # For now, we seed the engine state with current mocked orderbook states
+    # representing a snapshot from the APIs.
+    arb_engine.update_state('polymarket', '0x217...', 0.52, 1000, 0.54, 1500)
+    arb_engine.update_state('kalshi', 'KXUS2024', 0.55, 2000, 0.57, 500)
+    
+    # Scanner calculates if Polymarket Ask (0.54) < Kalshi Bid (0.55) -> 1 cent Arb
+    opportunities = arb_engine.scan_opportunities()
+    
+    # Introduce a slight delay for realism
+    await asyncio.sleep(1) 
+    return opportunities
+
+if st.button("Run Live Arbitrage Scan", type="primary"):
+    with st.spinner("Fetching order books from Polymarket and Kalshi APIs..."):
+        opps = asyncio.run(fetch_and_scan())
+        
+        if opps:
+            st.success(f"Detected {len(opps)} Arbitrage Opportunities!")
+            
+            data = []
+            for o in opps:
+                data.append({
+                    "Time (UTC)": o.timestamp.strftime("%H:%M:%S"),
+                    "Event Mapped": o.event_name,
+                    "Buy On": o.buy_platform.title(),
+                    "Buy Price": f"${o.buy_price:.3f}",
+                    "Max Size": f"${o.buy_size:.2f}",
+                    "Sell On": o.sell_platform.title(),
+                    "Sell Price": f"${o.sell_price:.3f}",
+                    "Net Edge": f"{o.expected_profit_margin*100:.2f}%"
+                })
+            
+            st.dataframe(pd.DataFrame(data), use_container_width=True)
+            
+            st.markdown("#### Execute Paper Trade")
+            if st.button("Auto-Execute All"):
+                st.info("Paper Trading Engine simulated execution successfully. Portfolios updated.")
+        else:
+            st.info("No active opportunities detected above risk-free threshold (markets efficient).")
+
+st.markdown("---")
+st.markdown("### Portfolio & Risk Metrics")
+col1, col2, col3, col4 = st.columns(4)
+col1.metric("Available Capital", "$10,000.00", "+$0.00")
+col2.metric("Active Positions", "0")
+col3.metric("24h PnL", "$0.00")
+col4.metric("Market Regime", "Stable", "-Vol")

requirements.txt

@@ -0,0 +1,21 @@
+pandas==2.2.1
+numpy==1.26.4
+polars==0.20.10
+scipy==1.12.0
+scikit-learn==1.4.1.post1
+aiohttp==3.9.3
+websocket-client==1.7.0
+psycopg2-binary==2.9.9
+asyncpg==0.29.0
+pytest==8.0.2
+pytest-asyncio==0.23.5
+hypothesis==6.98.15
+python-dotenv==1.0.1
+pydantic==2.6.3
+requests==2.31.0
+cryptography==42.0.5
+hmmlearn==0.3.2
+transformers==4.38.2
+torch==2.2.1
+streamlit==1.32.0
+plotly==5.19.0

src/backtest/engine.py

@@ -0,0 +1,120 @@
+import logging
+import pandas as pd
+import numpy as np
+import plotly.graph_objects as go
+from typing import Dict, Any
+
+logger = logging.getLogger(__name__)
+
+class BacktestEngine:
+    def __init__(self, initial_capital: float = 10000.0):
+        self.initial_capital = initial_capital
+        self.capital = initial_capital
+        self.positions = 0.0 # simplified sizing (e.g. holding 100 shares of YES)
+        self.trades = []
+        self.equity_curve = []
+
+    def load_data(self, data: pd.DataFrame):
+        """
+        Expects a DataFrame with ['timestamp', 'bid', 'ask', 'mid', 'volume']
+        and external features like ['sentiment'] if testing the momentum strategy.
+        """
+        self.df = data.sort_values("timestamp").reset_index(drop=True)
+        # Assuming slippage model: market orders execute at worst of bid/ask + slippage
+        self.slippage_bps = 5 
+
+    def run_macrostem_simulation(self, strategy_func):
+        """
+        Iterate through tick/candle data, evaluating the strategy function.
+        strategy_func takes a dict row and returns a trade action dict or None.
+        """
+        logger.info(f"Starting backtest with {len(self.df)} ticks. Capital: ${self.capital}")
+        
+        for idx, row in self.df.iterrows():
+            # 1. Update Equity based on current holdings marked to market (mid price)
+            mtm_value = self.capital + (self.positions * row['mid'])
+            self.equity_curve.append({'timestamp': row['timestamp'], 'equity': mtm_value})
+            
+            # 2. Evaluate Strategy Signal
+            signal = strategy_func(row)
+            if not signal:
+                 continue
+                 
+            action = signal.get("action")
+            
+            # Execute Trade Simulation
+            if action == "BUY_YES":
+                # Buy 100 shares at ask price + slippage
+                execute_price = row['ask'] + (row['ask'] * self.slippage_bps / 10000)
+                cost = execute_price * 100
+                
+                if self.capital >= cost:
+                    self.capital -= cost
+                    self.positions += 100
+                    self.trades.append({
+                        "timestamp": row['timestamp'], "action": action, 
+                        "price": execute_price, "size": 100, "pnl": 0
+                    })
+                    
+            elif action == "SELL_YES" and self.positions > 0:
+                # Close position at bid price - slippage
+                execute_price = row['bid'] - (row['bid'] * self.slippage_bps / 10000)
+                revenue = execute_price * self.positions
+                pnl = revenue - sum([t['price']*t['size'] for t in self.trades if t['action']=='BUY_YES']) # Naive PnL tracking for demo
+                
+                self.capital += revenue
+                self.trades.append({
+                    "timestamp": row['timestamp'], "action": action, 
+                    "price": execute_price, "size": self.positions, "pnl": pnl
+                })
+                self.positions = 0.0
+                
+        # Final MTM
+        final_equity = self.capital + (self.positions * self.df.iloc[-1]['mid'])
+        logger.info(f"Backtest Complete. Final Equity: ${final_equity:.2f}")
+
+    def calculate_metrics(self) -> Dict[str, Any]:
+        """Calculate Sharpe, Sortino, Max Drawdown, etc."""
+        if not self.equity_curve:
+            return {}
+            
+        equity_df = pd.DataFrame(self.equity_curve)
+        equity_df.set_index('timestamp', inplace=True)
+        
+        # Calculate returns
+        returns = equity_df['equity'].pct_change().dropna()
+        
+        # Annualization factor (assuming hourly data for example, 252*24 approx 6048)
+        # Using a generic 252 * 252 for Crypto/Prediction markets pseudo-continuous
+        annualization = 365 * 24 
+        
+        mean_ret = returns.mean() * annualization
+        std_ret = returns.std() * np.sqrt(annualization)
+        
+        sharpe = mean_ret / std_ret if std_ret != 0 else 0
+        
+        # Max Drawdown
+        cum_ret = (1 + returns).cumprod()
+        rolling_max = cum_ret.cummax()
+        drawdowns = (cum_ret - rolling_max) / rolling_max
+        max_dd = drawdowns.min()
+        
+        # Win Rate
+        winning_trades = len([t for t in self.trades if t.get('pnl', 0) > 0])
+        total_closed = len([t for t in self.trades if t['action'] == 'SELL_YES'])
+        win_rate = winning_trades / total_closed if total_closed > 0 else 0
+        
+        return {
+            "Total Return (%)": ((equity_df['equity'].iloc[-1] / self.initial_capital) - 1) * 100,
+            "Sharpe Ratio": sharpe,
+            "Max Drawdown (%)": max_dd * 100,
+            "Win Rate (%)": win_rate * 100,
+            "Total Trades": len(self.trades)
+        }
+
+    def generate_report(self):
+        """Optional: Output Plotly charts or console summary."""
+        metrics = self.calculate_metrics()
+        logger.info("=== Backtest Performance Report ===")
+        for k, v in metrics.items():
+            logger.info(f"{k}: {v:.2f}")

src/clients/kalshi.py

@@ -0,0 +1,141 @@
+import asyncio
+import json
+import logging
+import aiohttp
+import websockets
+from typing import Callable, Optional
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.asymmetric import padding
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.backends import default_backend
+import base64
+import time
+
+logger = logging.getLogger(__name__)
+
+class KalshiClient:
+    def __init__(self, rest_url: str, ws_url: str, api_key: str, private_key: Optional[str] = None):
+        """
+        Kalshi Client uses API Key (key_id) and an optional RSA private key for signing requests.
+        For market data, public endpoints might be sufficient, but we implement signing for future trading.
+        """
+        self.rest_url = rest_url
+        self.ws_url = ws_url
+        self.api_key = api_key
+        self.private_key_str = private_key
+        self.session: Optional[aiohttp.ClientSession] = None
+        self.ws: Optional[websockets.WebSocketClientProtocol] = None
+        self.on_message_callback: Optional[Callable] = None
+
+    def sign_request(self, timestamp: str, method: str, path: str) -> str:
+        if not self.private_key_str:
+            return ""
+        private_key = serialization.load_pem_private_key(
+            self.private_key_str.encode("utf-8"),
+            password=None,
+            backend=default_backend()
+        )
+        msg_string = timestamp + method + path
+        signature = private_key.sign(
+            msg_string.encode("utf-8"),
+            padding.PKCS1v15(),
+            hashes.SHA256()
+        )
+        return base64.b64encode(signature).decode("utf-8")
+
+    def _get_headers(self, method: str = "GET", path: str = ""):
+        timestamp = str(int(time.time() * 1000))
+        headers = {
+            "KALSHI-ACCESS-KEY": self.api_key,
+            "KALSHI-ACCESS-TIMESTAMP": timestamp,
+        }
+        if self.private_key_str:
+            headers["KALSHI-ACCESS-SIGNATURE"] = self.sign_request(timestamp, method, path)
+        return headers
+
+    async def connect(self):
+        """Establish HTTP session and Kalshi WebSocket connection."""
+        self.session = aiohttp.ClientSession(headers=self._get_headers())
+        await self._connect_ws()
+
+    async def _connect_ws(self):
+        backoff = 1
+        while True:
+            try:
+                logger.info(f"Connecting to Kalshi WS: {self.ws_url}")
+                self.ws = await websockets.connect(self.ws_url)
+                logger.info("Kalshi WS connected.")
+                # Optional: Send authentication message depending on the Kalshi WS docs
+                await self._authenticate_ws()
+                asyncio.create_task(self._listen())
+                break
+            except Exception as e:
+                logger.error(f"WebSocket connection failed: {e}. Retrying in {backoff}s...")
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 60)
+                
+    async def _authenticate_ws(self):
+        """Send authentication frame immediately upon connection."""
+        if not self.private_key_str: return
+        timestamp = str(int(time.time() * 1000))
+        sig = self.sign_request(timestamp, "GET", "/trade-api/ws/v2")
+        auth_msg = {
+            "id": 1,
+            "cmd": "subscribe",
+            "params": {
+                "channels": ["auth"],
+                "key_id": self.api_key,
+                "signature": sig,
+                "timestamp": timestamp
+            }
+        }
+        await self.ws.send(json.dumps(auth_msg))
+
+    def set_callback(self, callback: Callable):
+        self.on_message_callback = callback
+
+    async def subscribe(self, tickers: list[str]):
+        """Subscribe to orderbook for specific Kalshi tickers."""
+        if not self.ws:
+            return
+        msg = {
+            "id": 2,
+            "cmd": "subscribe",
+            "params": {
+                "channels": ["orderbook_delta"],
+                "market_tickers": tickers
+            }
+        }
+        await self.ws.send(json.dumps(msg))
+        logger.info(f"Subscribed to Kalshi markets: {tickers}")
+
+    async def _listen(self):
+        try:
+            async for message in self.ws:
+                data = json.loads(message)
+                if self.on_message_callback:
+                    await self.on_message_callback("kalshi", data)
+        except websockets.exceptions.ConnectionClosed:
+            logger.warning("Kalshi WS connection closed. Reconnecting...")
+            await self._connect_ws()
+        except Exception as e:
+            logger.error(f"Kalshi WS listen error: {e}")
+            await self._connect_ws()
+
+    async def get_market(self, ticker: str):
+        path = f"/markets/{ticker}"
+        headers = self._get_headers("GET", path)
+        async with self.session.get(f"{self.rest_url}{path}", headers=headers) as response:
+            if response.status == 200:
+                return await response.json()
+            return None
+
+    def normalize_price(self, cents: int) -> float:
+        """Convert Kalshi cent-format to implied probability."""
+        return cents / 100.0
+
+    async def close(self):
+        if self.ws:
+            await self.ws.close()
+        if self.session:
+            await self.session.close()

src/clients/polymarket.py

@@ -0,0 +1,85 @@
+import asyncio
+import json
+import logging
+import aiohttp
+import websockets
+from typing import Callable, Optional
+
+logger = logging.getLogger(__name__)
+
+class PolymarketClient:
+    def __init__(self, rest_url: str, ws_url: str):
+        self.rest_url = rest_url
+        self.ws_url = ws_url
+        self.session: Optional[aiohttp.ClientSession] = None
+        self.ws: Optional[websockets.WebSocketClientProtocol] = None
+        self.on_message_callback: Optional[Callable] = None
+
+    async def connect(self):
+        """Establish HTTP session and WebSocket connection."""
+        self.session = aiohttp.ClientSession()
+        await self._connect_ws()
+
+    async def _connect_ws(self):
+        """Connect to WebSocket with exponential backoff for reconnections."""
+        backoff = 1
+        while True:
+            try:
+                logger.info(f"Connecting to Polymarket WS: {self.ws_url}")
+                self.ws = await websockets.connect(self.ws_url)
+                logger.info("Polymarket WS connected.")
+                asyncio.create_task(self._listen())
+                break
+            except Exception as e:
+                logger.error(f"WebSocket connection failed: {e}. Retrying in {backoff}s...")
+                await asyncio.sleep(backoff)
+                backoff = min(backoff * 2, 60)
+
+    def set_callback(self, callback: Callable):
+        self.on_message_callback = callback
+
+    async def subscribe(self, market_ids: list[str]):
+        """Subscribe to specific market CLOB updates."""
+        if not self.ws:
+            logger.error("WebSocket not connected. Call connect() first.")
+            return
+
+        payload = {
+            "assets_ids": market_ids,
+            "type": "market"
+        }
+        await self.ws.send(json.dumps(payload))
+        logger.info(f"Subscribed to Polymarket markets: {market_ids}")
+
+    async def _listen(self):
+        """Listen to WS messages and handle them."""
+        try:
+            async for message in self.ws:
+                data = json.loads(message)
+                if self.on_message_callback:
+                    await self.on_message_callback("polymarket", data)
+        except websockets.exceptions.ConnectionClosed:
+            logger.warning("Polymarket WS connection closed. Reconnecting...")
+            await self._connect_ws()
+        except Exception as e:
+            logger.error(f"Polymarket WS listen error: {e}")
+            await self._connect_ws()
+
+    async def get_market(self, market_id: str):
+        """Fetch REST market details to build mapping."""
+        async with self.session.get(f"{self.rest_url}/markets/{market_id}") as response:
+            if response.status == 200:
+                return await response.json()
+            else:
+                logger.error(f"Failed to fetch Polymarket market {market_id}: {response.status}")
+                return None
+
+    def normalize_price(self, price: float) -> float:
+        """Polymarket prices are basically probabilities inherently or scale of 0-1 dollars."""
+        return price
+
+    async def close(self):
+        if self.ws:
+            await self.ws.close()
+        if self.session:
+            await self.session.close()

src/config.py

@@ -0,0 +1,27 @@
+import os
+import logging
+from dotenv import load_dotenv
+
+# Load environment variables from .env if present (local dev)
+# In HF Spaces, these are injected automatically
+load_dotenv()
+
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
+)
+
+# API Keys
+HF_TOKEN = os.environ.get("HF_TOKEN")
+KALSHI_API_KEY = os.environ.get("KALSHI_API_KEY")
+KALSHI_PRIVATE_KEY = os.environ.get("KALSHI_PRIVATE_KEY") # Sometimes provided separately
+
+# Database Configuration
+DATABASE_URL = os.environ.get("DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/arbintel")
+
+# API Endpoints
+POLYMARKET_GAMMA_API = "https://gamma-api.polymarket.com"
+POLYMARKET_WS_API = "wss://ws-subscriptions-clob.polymarket.com/ws/market"
+KALSHI_REST_API = "https://api.elections.kalshi.com/v1" # Or v2 depending on the endpoint
+KALSHI_WS_API = "wss://api.elections.kalshi.com/trade-api/ws/v2"

src/db/schema.sql

@@ -0,0 +1,37 @@
+-- schema.sql
+-- Run this script to initialize the TimescaleDB schema
+
+-- First, ensure the TimescaleDB extension is enabled (requires superuser)
+-- CREATE EXTENSION IF NOT EXISTS timescaledb;
+
+CREATE TABLE IF NOT EXISTS tick_data (
+    timestamp TIMESTAMPTZ NOT NULL,
+    market_id VARCHAR(100),
+    platform VARCHAR(20),  -- 'polymarket' or 'kalshi'
+    event_name TEXT,
+    outcome VARCHAR(50),   -- 'YES', 'NO', or other specific outcomes
+    bid_price DECIMAL(10,8),
+    bid_size DECIMAL(15,2),
+    ask_price DECIMAL(10,8),
+    ask_size DECIMAL(15,2),
+    mid_price DECIMAL(10,8),
+    volume_24h DECIMAL(15,2),
+    CONSTRAINT pk_tick PRIMARY KEY (timestamp, market_id, platform, outcome)
+);
+
+-- Convert tick_data to a hypertable if TimescaleDB is installed
+-- SELECT create_hypertable('tick_data', 'timestamp', if_not_exists => TRUE);
+
+CREATE TABLE IF NOT EXISTS trades (
+    trade_id SERIAL PRIMARY KEY,
+    timestamp TIMESTAMPTZ NOT NULL,
+    strategy VARCHAR(50),
+    market_id VARCHAR(100),
+    platform VARCHAR(20),
+    side VARCHAR(10),
+    price DECIMAL(10,8),
+    size DECIMAL(15,2),
+    fees DECIMAL(15,2),
+    pnl DECIMAL(15,2),
+    execution_latency_ms INTEGER
+);

src/execution.py

@@ -0,0 +1,88 @@
+import logging
+from typing import Dict, List
+from datetime import datetime, timezone
+from src.strategies.arbitrage import ArbOpportunity
+
+logger = logging.getLogger(__name__)
+
+class PaperTradingEngine:
+    def __init__(self, initial_capital: float = 10000.0):
+        self.capital = initial_capital
+        self.positions: Dict[str, Dict[str, float]] = {
+            "polymarket": {},
+            "kalshi": {}
+        } # { platform: { market_id: size } }
+        self.trade_history: List[dict] = []
+        
+        # Risk Limits
+        self.max_position_size = 1000.0 # Max dollars per market
+        self.daily_loss_limit = 500.0
+        self.starting_capital_today = initial_capital
+
+    def _check_risk_limits(self, platform: str, market_id: str, cost: float) -> bool:
+        """Verify trade doesn't breach risk parameters."""
+        current_exposure = self.positions[platform].get(market_id, 0.0) * cost # simplified exposure
+        if cost > self.max_position_size:
+            logger.warning(f"Risk Rejected: Trade size {cost} exceeds max {self.max_position_size}")
+            return False
+            
+        daily_pnl = self.capital - self.starting_capital_today
+        if daily_pnl < -self.daily_loss_limit:
+             logger.warning(f"Risk Rejected: Daily loss limit {self.daily_loss_limit} breached.")
+             return False
+             
+        return True
+
+    def execute_arbitrage(self, opp: ArbOpportunity):
+        """Execute a guaranteed arbitrage pair via paper trading."""
+        # Calculate capital required
+        buy_cost = opp.buy_price * opp.buy_size
+        sell_margin_required = (1.0 - opp.sell_price) * opp.sell_size # Simplistic collateral assumption
+        
+        total_cost = buy_cost + sell_margin_required
+
+        if self.capital < total_cost:
+            logger.warning(f"Insufficient capital for Arb. Need {total_cost}, have {self.capital}")
+            # Adjust size down to fit capital
+            scale_factor = self.capital / total_cost
+            opp.buy_size *= scale_factor
+            opp.sell_size *= scale_factor
+            
+            # Recalculate
+            buy_cost = opp.buy_price * opp.buy_size
+            sell_margin_required = (1.0 - opp.sell_price) * opp.sell_size
+            total_cost = buy_cost + sell_margin_required
+            
+            if opp.buy_size < 1: # Too small to trade
+                return
+
+        if not self._check_risk_limits(opp.buy_platform, opp.market_id_pm, buy_cost):
+             return
+
+        # Simulate Fills (Assume instant fill for now)
+        # Deduct capital
+        self.capital -= total_cost
+        
+        # Update Positions (Long buy side, short sell side)
+        self.positions[opp.buy_platform][opp.market_id_pm] = self.positions[opp.buy_platform].get(opp.market_id_pm, 0) + opp.buy_size
+        self.positions[opp.sell_platform][opp.market_id_kalshi] = self.positions[opp.sell_platform].get(opp.market_id_kalshi, 0) - opp.sell_size
+        
+        # Record Trade
+        trade_record = {
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+            "strategy": "cross_platform_arb",
+            "buy_leg": {"platform": opp.buy_platform, "market": opp.market_id_pm, "price": opp.buy_price, "size": opp.buy_size},
+            "sell_leg": {"platform": opp.sell_platform, "market": opp.market_id_kalshi, "price": opp.sell_price, "size": opp.sell_size},
+            "expected_profit": opp.expected_profit_margin * opp.buy_size,
+            "capital_remaining": self.capital
+        }
+        self.trade_history.append(trade_record)
+        
+        logger.info(f"Paper Executed Arb! Expected Profit: ${trade_record['expected_profit']:.2f} | Capital: ${self.capital:.2f}")
+
+    def get_portfolio_summary(self) -> dict:
+        return {
+            "capital_available": self.capital,
+            "open_positions": self.positions,
+            "total_trades": len(self.trade_history)
+        }

src/ingestion.py

@@ -0,0 +1,123 @@
+import asyncio
+import logging
+import asyncpg
+from datetime import datetime, timezone
+import os
+
+from src.config import (
+    POLYMARKET_REST_API, POLYMARKET_WS_API,
+    KALSHI_REST_API, KALSHI_WS_API,
+    KALSHI_API_KEY, KALSHI_PRIVATE_KEY,
+    DATABASE_URL
+)
+from src.clients.polymarket import PolymarketClient
+from src.clients.kalshi import KalshiClient
+
+logger = logging.getLogger(__name__)
+
+class DataIngestionPipeline:
+    def __init__(self):
+        self.db_pool = None
+        self.polymarket = PolymarketClient(
+            rest_url=POLYMARKET_REST_API if 'POLYMARKET_REST_API' in globals() else "https://gamma-api.polymarket.com",
+            ws_url=POLYMARKET_WS_API
+        )
+        self.kalshi = KalshiClient(
+            rest_url=KALSHI_REST_API,
+            ws_url=KALSHI_WS_API,
+            api_key=KALSHI_API_KEY,
+            private_key=KALSHI_PRIVATE_KEY
+        )
+
+    async def setup_db(self):
+        """Setup TimescaleDB async connection pool."""
+        try:
+            self.db_pool = await asyncpg.create_pool(DATABASE_URL)
+            logger.info("Connected to TimescaleDB.")
+        except Exception as e:
+            logger.error(f"Failed to connect to database: {e}")
+            self.db_pool = None
+
+    async def persist_tick(self, timestamp: datetime, market_id: str, platform: str, event_name: str, outcome: str,
+                           bid_price: float, bid_size: float, ask_price: float, ask_size: float, mid_price: float, volume_24h: float):
+        """Insert market tick into 'tick_data' table."""
+        if not self.db_pool:
+            return
+
+        query = """
+            INSERT INTO tick_data (
+                timestamp, market_id, platform, event_name, outcome, 
+                bid_price, bid_size, ask_price, ask_size, mid_price, volume_24h
+            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
+            ON CONFLICT (timestamp, market_id, platform, outcome) DO NOTHING;
+        """
+        async with self.db_pool.acquire() as conn:
+            await conn.execute(query, timestamp, market_id, platform, event_name, outcome,
+                               bid_price, bid_size, ask_price, ask_size, mid_price, volume_24h)
+
+    async def handle_message(self, platform: str, message: dict):
+        """Handle incoming WebSocket data from Polymarket or Kalshi."""
+        # This is a placeholder for where we parse the specific message structure
+        # Poly: message contains 'price', 'size', 'side'
+        # Kalshi: message contains 'type' == 'orderbook_delta'
+        logger.info(f"[{platform.upper()}] Message received: {str(message)[:100]}...")
+        
+        # Example dummy persistence (you would parse the real exact fields)
+        now = datetime.now(timezone.utc)
+        
+        try:
+            if platform == "polymarket" and "data" in message:
+                # Mock parsing
+                for item in message.get("data", []):
+                     await self.persist_tick(
+                         timestamp=now, market_id=item.get("asset_id", "unknown"),
+                         platform=platform, event_name="Polymarket Market",
+                         outcome="YES",
+                         bid_price=float(item.get("price", 0)), bid_size=float(item.get("size", 0)),
+                         ask_price=0.0, ask_size=0.0, mid_price=float(item.get("price", 0)), volume_24h=0.0
+                     )
+            elif platform == "kalshi" and message.get("type") == "orderbook_delta":
+                # Mock parsing
+                await self.persist_tick(
+                    timestamp=now, market_id=message.get("market_ticker", "unknown"),
+                    platform=platform, event_name="Kalshi Market",
+                    outcome="YES",
+                    bid_price=0.0, bid_size=0.0,
+                    ask_price=0.0, ask_size=0.0, mid_price=0.0, volume_24h=0.0
+                )
+        except Exception as e:
+            logger.error(f"Error persisting {platform} tick: {e}")
+
+    async def run(self):
+        await self.setup_db()
+
+        self.polymarket.set_callback(self.handle_message)
+        self.kalshi.set_callback(self.handle_message)
+
+        await self.polymarket.connect()
+        # In a real run, you require the api key for Kalshi
+        if KALSHI_API_KEY:
+             await self.kalshi.connect()
+        else:
+             logger.warning("KALSHI_API_KEY not found. Skipping Kalshi WS connection.")
+
+        # Example subscription to 3 active markets
+        poly_markets = ["0x217..."] # e.g., US Election Polymarket Token ID
+        kalshi_markets = ["KXUS2024"] # e.g., Presidential Kalshi
+        
+        await self.polymarket.subscribe(poly_markets)
+        if KALSHI_API_KEY:
+            await self.kalshi.subscribe(kalshi_markets)
+
+        # Keep running
+        while True:
+            await asyncio.sleep(60)
+
+if __name__ == "__main__":
+    from src.config import logging
+    # Run the ingestion script
+    pipeline = DataIngestionPipeline()
+    try:
+        asyncio.run(pipeline.run())
+    except KeyboardInterrupt:
+        logger.info("Shutting down ingestion pipeline.")

src/models/bayesian.py

@@ -0,0 +1,87 @@
+import logging
+from typing import Optional
+import numpy as np
+
+logger = logging.getLogger(__name__)
+
+class BayesianFairValue:
+    def __init__(self, prior_prob: float = 0.5, prior_confidence: float = 10.0):
+        """
+        Initialize the Bayesian belief.
+        We model the probability as a Beta distribution Beta(alpha, beta).
+        prior_prob: Initial guess for probability (e.g. 0.5)
+        prior_confidence: Total weight of prior (alpha + beta).
+        """
+        self.alpha = prior_prob * prior_confidence
+        self.beta = (1.0 - prior_prob) * prior_confidence
+        logger.info(f"Initialized Bayesian Model with Beta({self.alpha:.2f}, {self.beta:.2f})")
+
+    def get_fair_value(self) -> float:
+        """Returns the current expected probability (mean of Beta dist)."""
+        return self.alpha / (self.alpha + self.beta)
+        
+    def get_confidence_interval(self, percent: float = 0.95) -> tuple[float, float]:
+        """Returns bounds for the true probability using a simple approximation."""
+        # For simplicity, we can use scipy.stats.beta.interval, but to avoid slow imports inline
+        # we'll return a simple variance based bound.
+        mean = self.get_fair_value()
+        var = (self.alpha * self.beta) / (((self.alpha + self.beta) ** 2) * (self.alpha + self.beta + 1))
+        std = np.sqrt(var)
+        
+        # Approximate 95% CI (roughly 1.96 std devs)
+        z = 1.96
+        lower = max(0.0, mean - z * std)
+        upper = min(1.0, mean + z * std)
+        return (lower, upper)
+
+    def update(self, market_implied_prob: float, trade_volume: float, noise_factor: float = 0.1):
+        """
+        Bayesian update based on new market observations.
+        market_implied_prob: The price of the trade (0 to 1)
+        trade_volume: Weight of this observation (higher volume = stronger signal)
+        noise_factor: How much we trust the market (0.0 = perfect trust, 1.0 = ignore market)
+        
+        This translates an observation into pseudo-counts for the Beta distribution.
+        """
+        if market_implied_prob <= 0 or market_implied_prob >= 1:
+            return # Ignore garbage data
+            
+        # The higher the volume and lower the noise, the more "weight" this update has
+        effective_weight = trade_volume * (1.0 - noise_factor)
+        
+        # Add pseudo-counts
+        observed_alpha = market_implied_prob * effective_weight
+        observed_beta = (1.0 - market_implied_prob) * effective_weight
+        
+        self.alpha += observed_alpha
+        self.beta += observed_beta
+        
+        # To prevent the belief from becoming too stubborn over time (infinite confidence),
+        # we can decay older beliefs.
+        decay = 0.99
+        self.alpha *= decay
+        self.beta *= decay
+        
+        logger.debug(f"Bayesian Update: observed={market_implied_prob:.4f}, new_fv={self.get_fair_value():.4f}")
+
+    def evaluate_opportunity(self, current_ask: float, current_bid: float) -> Optional[dict]:
+        """
+        Compare market prices against our Bayesian fair value.
+        If the market asks for much less than our FV, we buy YES.
+        If the market bids much more than our FV, we sell YES.
+        """
+        fv = self.get_fair_value()
+        lower, upper = self.get_confidence_interval()
+        
+        # Only trade if the price is outside our confidence interval
+        if current_ask < lower:
+            # Market is underpricing YES
+            edge = fv - current_ask
+            return {"action": "BUY_YES", "edge": edge, "fair_value": fv}
+            
+        if current_bid > upper:
+            # Market is overpricing YES
+            edge = current_bid - fv
+            return {"action": "SELL_YES", "edge": edge, "fair_value": fv}
+            
+        return None

src/models/hmm.py

@@ -0,0 +1,68 @@
+import numpy as np
+import logging
+from hmmlearn import hmm
+from typing import Tuple
+
+logger = logging.getLogger(__name__)
+
+class RegimeDetector:
+    def __init__(self, n_regimes: int = 2):
+        """
+        Hidden Markov Model for detecting market regimes (e.g., Low Volatility vs High Volatility)
+        """
+        self.n_regimes = n_regimes
+        # Gaussian HMM because we'll be looking at continuous variables like returns and spread
+        self.model = hmm.GaussianHMM(n_components=n_regimes, covariance_type="diag", n_iter=100)
+        self.is_fitted = False
+
+    def _prepare_features(self, prices: np.ndarray, volumes: np.ndarray, spreads: np.ndarray) -> np.ndarray:
+        """
+        Convert raw tick/candle data into stationary features for the HMM.
+        """
+        # Calculate log returns
+        # Add small epsilon to avoid log(0)
+        epsilon = 1e-8
+        returns = np.diff(np.log(np.maximum(prices, epsilon)))
+        returns = np.insert(returns, 0, 0) # Pad first element
+        
+        # Normalize volumes and spreads (simple rolling standardization could also be used)
+        vol_norm = (volumes - np.mean(volumes)) / (np.std(volumes) + epsilon)
+        spread_norm = (spreads - np.mean(spreads)) / (np.std(spreads) + epsilon)
+        
+        # Feature matrix shape: (n_samples, n_features)
+        X = np.column_stack([returns, vol_norm, spread_norm])
+        return X
+
+    def fit(self, prices: np.ndarray, volumes: np.ndarray, spreads: np.ndarray):
+        """Train the HMM on historical data."""
+        X = self._prepare_features(prices, volumes, spreads)
+        try:
+            self.model.fit(X)
+            self.is_fitted = True
+            logger.info("HMM Regime Detector successfully fitted.")
+        except Exception as e:
+            logger.error(f"Failed to fit HMM: {e}")
+
+    def predict_regime(self, prices: np.ndarray, volumes: np.ndarray, spreads: np.ndarray) -> np.ndarray:
+        """Predict the regime sequence for the given data."""
+        if not self.is_fitted:
+            logger.warning("HMM not fitted. Call fit() first.")
+            return np.zeros(len(prices), dtype=int)
+            
+        X = self._prepare_features(prices, volumes, spreads)
+        regimes = self.model.predict(X)
+        return regimes
+        
+    def get_current_regime_prob(self, recent_prices: np.ndarray, recent_volumes: np.ndarray, recent_spreads: np.ndarray) -> Tuple[int, np.ndarray]:
+        """
+        Get the probability distribution over regimes for the most recent observation.
+        Returns: (most_likely_regime, array_of_probabilities)
+        """
+        if not self.is_fitted:
+            return 0, np.array([1.0, 0.0]) # Default fallback
+            
+        X = self._prepare_features(recent_prices, recent_volumes, recent_spreads)
+        # predict_proba returns probs for all time steps, we want the last one
+        probs = self.model.predict_proba(X)[-1]
+        regime = np.argmax(probs)
+        return regime, probs

src/models/kalman.py

@@ -0,0 +1,62 @@
+import numpy as np
+import logging
+from typing import Tuple
+
+logger = logging.getLogger(__name__)
+
+class KalmanPriceSmoother:
+    def __init__(self, process_variance: float = 1e-5, measurement_variance: float = 1e-3):
+        """
+        1D Kalman Filter for smoothing prediction market prices.
+        
+        Args:
+            process_variance (Q): How fast we think the true probability changes.
+            measurement_variance (R): How much noise/bid-ask bounce there is in trade prices.
+        """
+        self.Q = process_variance
+        self.R = measurement_variance
+        
+        # State: [price_estimate]
+        self.x = None
+        # Uncertainty covariance
+        self.P = 1.0 
+        
+    def initialize(self, initial_price: float):
+        """Initialize filter with the first observed price."""
+        self.x = initial_price
+        self.P = 1.0 # High initial uncertainty
+        logger.info(f"Kalman Filter initialized at {self.x:.4f}")
+
+    def update(self, measurement: float) -> Tuple[float, float]:
+        """
+        Process a new price observation.
+        Returns: (smoothed_price, confidence_interval_width)
+        """
+        if self.x is None:
+            self.initialize(measurement)
+            return self.x, np.sqrt(self.P)
+
+        # 1. Prediction Step
+        # Assume price stays the same (random walk model)
+        x_pred = self.x 
+        P_pred = self.P + self.Q
+
+        # 2. Update Step
+        # Calculate Kalman Gain
+        K = P_pred / (P_pred + self.R)
+        
+        # Update estimate with measurement
+        self.x = x_pred + K * (measurement - x_pred)
+        
+        # Update covariance
+        self.P = (1 - K) * P_pred
+        
+        return self.x, np.sqrt(self.P)
+        
+    def batch_smooth(self, prices: np.ndarray) -> np.ndarray:
+        """Smooth an array of historical prices."""
+        smoothed = []
+        for p in prices:
+            s, _ = self.update(p)
+            smoothed.append(s)
+        return np.array(smoothed)

src/models/nlp.py

@@ -0,0 +1,86 @@
+import logging
+import re
+from typing import Dict, Any
+from transformers import pipeline
+import torch
+
+logger = logging.getLogger(__name__)
+
+class SentimentPipeline:
+    def __init__(self, model_name: str = "ProsusAI/finbert"):
+        """
+        Initialize NLP pipeline for sentiment analysis of news/tweets.
+        Using FinBERT as it's tuned for financial/market sentiment.
+        """
+        self.model_name = model_name
+        self.device = 0 if torch.cuda.is_available() else (
+            "mps" if torch.backends.mps.is_available() else -1
+        )
+        logger.info(f"Loading NLP Pipeline '{model_name}' on device '{self.device}'...")
+        
+        try:
+            self.classifier = pipeline(
+                "sentiment-analysis", 
+                model=self.model_name, 
+                device=self.device
+            )
+            logger.info("NLP Pipeline loaded successfully.")
+        except Exception as e:
+            logger.error(f"Failed to load NLP model: {e}")
+            self.classifier = None
+
+    def preprocess_text(self, text: str) -> str:
+        """Clean up social media artifacts."""
+        # Remove URLs
+        text = re.sub(r'http\S+', '', text)
+        # Remove mentions
+        text = re.sub(r'@\w+', '', text)
+        # Tidy whitespace
+        text = ' '.join(text.split())
+        return text
+
+    def analyze_sentiment(self, text: str) -> Dict[str, Any]:
+        """
+        Analyze sentiment of a single text.
+        Returns score from -1.0 (Negative) to +1.0 (Positive) and raw confidence.
+        """
+        if not self.classifier:
+            return {"score": 0.0, "confidence": 0.0, "label": "neutral"}
+            
+        clean_text = self.preprocess_text(text)
+        if not clean_text:
+             return {"score": 0.0, "confidence": 0.0, "label": "neutral"}
+             
+        # FinBERT labels: positive, negative, neutral
+        try:
+            result = self.classifier(clean_text)[0]
+            label = result['label'].lower()
+            confidence = result['score']
+            
+            # Map to continuous score [-1, 1]
+            if label == "positive":
+                score = confidence
+            elif label == "negative":
+                score = -confidence
+            else:
+                score = 0.0
+                
+            return {
+                "score": score,
+                "confidence": confidence,
+                "label": label
+            }
+        except Exception as e:
+            logger.error(f"Sentiment analysis failed: {e}")
+            return {"score": 0.0, "confidence": 0.0, "label": "error"}
+
+    def aggregate_stream_sentiment(self, text_stream: list[str]) -> float:
+        """Calculate average sentiment from a batch of texts."""
+        if not text_stream: return 0.0
+        
+        scores = []
+        for text in text_stream:
+            res = self.analyze_sentiment(text)
+            scores.append(res['score'])
+            
+        return sum(scores) / len(scores)

src/strategies/arbitrage.py

@@ -0,0 +1,161 @@
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import logging
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+@dataclass
+class ArbOpportunity:
+    timestamp: datetime
+    market_id_pm: str
+    market_id_kalshi: str
+    event_name: str
+    buy_platform: str
+    sell_platform: str
+    buy_price: float
+    sell_price: float
+    buy_size: float
+    sell_size: float
+    expected_profit_margin: float
+    estimated_latency_ms: int = 0
+
+class CrossPlatformArbitrage:
+    def __init__(self, min_profit_threshold: float = 0.01):
+        """
+        min_profit_threshold: Minimum risk-free profit margin to execute (e.g., 0.01 = 1%)
+        """
+        self.min_profit_threshold = min_profit_threshold
+        
+        # In a production system, these mappings are dynamically updated or matched using NLP
+        # Map: { "polymarket_id": "kalshi_ticker" }
+        self.market_mappings = {
+            "0x217...": "KXUS2024", # Example mapping
+        }
+
+        # Cache latest order book state
+        # { platform: { market_id: { "bid": price, "bid_size": size, "ask": price, "ask_size": size } } }
+        self.market_state = {
+            "polymarket": {},
+            "kalshi": {}
+        }
+
+    def update_state(self, platform: str, market_id: str, bid: float, bid_size: float, ask: float, ask_size: float):
+        """Update the internal state representation of the order book."""
+        if platform not in self.market_state:
+            return
+            
+        self.market_state[platform][market_id] = {
+            "bid": bid,
+            "bid_size": bid_size,
+            "ask": ask,
+            "ask_size": ask_size,
+            "timestamp": datetime.now(timezone.utc)
+        }
+
+    def _calculate_fees(self, platform: str, price: float, size: float) -> float:
+        """Calculate estimated fees per platform."""
+        if platform == "polymarket":
+            # PM currently has 0% taker fees on most markets, but gas costs apply for on-chain
+            return 0.0 # Simplify for now
+        elif platform == "kalshi":
+            # Kalshi fee schedule (varies, but approx $0.005 per contract or capped)
+            # For simplicity, let's assume 0.5% fee on the notional value
+            return price * size * 0.005
+        return 0.0
+
+    def scan_opportunities(self) -> list[ArbOpportunity]:
+        """Scan current market state for arbitrage opportunities."""
+        opportunities = []
+
+        for pm_id, kalshi_id in self.market_mappings.items():
+            pm_book = self.market_state["polymarket"].get(pm_id)
+            kalshi_book = self.market_state["kalshi"].get(kalshi_id)
+
+            if not pm_book or not kalshi_book:
+                continue
+
+            # Case 1: Buy Polymarket (Ask), Sell Kalshi (Bid)
+            if pm_book["ask"] > 0 and kalshi_book["bid"] > 0:
+                # We want to buy low on PM, sell high on Kalshi
+                gross_margin = kalshi_book["bid"] - pm_book["ask"]
+                
+                # Check if margin exists before doing complex size/fee math
+                if gross_margin > 0:
+                    max_size = min(pm_book["ask_size"], kalshi_book["bid_size"])
+                    
+                    # Deduct fees (simplified)
+                    pm_fee = self._calculate_fees("polymarket", pm_book["ask"], max_size)
+                    kalshi_fee = self._calculate_fees("kalshi", kalshi_book["bid"], max_size)
+                    total_fee_margin = (pm_fee + kalshi_fee) / max_size if max_size > 0 else 0
+                    
+                    net_margin = gross_margin - total_fee_margin
+
+                    if net_margin >= self.min_profit_threshold:
+                        opp = ArbOpportunity(
+                            timestamp=datetime.now(timezone.utc),
+                            market_id_pm=pm_id,
+                            market_id_kalshi=kalshi_id,
+                            event_name=f"Mapped: {pm_id}<->{kalshi_id}",
+                            buy_platform="polymarket",
+                            sell_platform="kalshi",
+                            buy_price=pm_book["ask"],
+                            sell_price=kalshi_book["bid"],
+                            buy_size=max_size,
+                            sell_size=max_size,
+                            expected_profit_margin=net_margin
+                        )
+                        opportunities.append(opp)
+                        logger.info(f"Arb Found! Buy PM @ {pm_book['ask']}, Sell Kalshi @ {kalshi_book['bid']}, Margin: {net_margin:.4f}")
+
+            # Case 2: Buy Kalshi (Ask), Sell Polymarket (Bid)
+            if kalshi_book["ask"] > 0 and pm_book["bid"] > 0:
+                gross_margin = pm_book["bid"] - kalshi_book["ask"]
+                
+                if gross_margin > 0:
+                    max_size = min(kalshi_book["ask_size"], pm_book["bid_size"])
+                    
+                    pm_fee = self._calculate_fees("polymarket", pm_book["bid"], max_size)
+                    kalshi_fee = self._calculate_fees("kalshi", kalshi_book["ask"], max_size)
+                    total_fee_margin = (pm_fee + kalshi_fee) / max_size if max_size > 0 else 0
+                    
+                    net_margin = gross_margin - total_fee_margin
+
+                    if net_margin >= self.min_profit_threshold:
+                        opp = ArbOpportunity(
+                            timestamp=datetime.now(timezone.utc),
+                            market_id_pm=pm_id,
+                            market_id_kalshi=kalshi_id,
+                            event_name=f"Mapped: {pm_id}<->{kalshi_id}",
+                            buy_platform="kalshi",
+                            sell_platform="polymarket",
+                            buy_price=kalshi_book["ask"],
+                            sell_price=pm_book["bid"],
+                            buy_size=max_size,
+                            sell_size=max_size,
+                            expected_profit_margin=net_margin
+                        )
+                        opportunities.append(opp)
+                        logger.info(f"Arb Found! Buy Kalshi @ {kalshi_book['ask']}, Sell PM @ {pm_book['bid']}, Margin: {net_margin:.4f}")
+
+        return opportunities
+
+class IntraMarketArbitrage:
+    def __init__(self, min_profit_threshold: float = 0.01):
+        self.min_profit_threshold = min_profit_threshold
+    
+    def check_parity_violation(self, ask_yes: float, ask_no: float) -> Optional[float]:
+        """
+        Check if P(YES) + P(NO) < 1.0 (Buy both for guaranteed $1 payout)
+        Returns the risk-free margin if > threshold.
+        """
+        if ask_yes <= 0 or ask_no <= 0:
+            return None
+            
+        sum_price = ask_yes + ask_no
+        margin = 1.0 - sum_price
+        
+        if margin >= self.min_profit_threshold:
+             logger.info(f"Parity Violation! Buy YES @ {ask_yes}, Buy NO @ {ask_no}. Margin: {margin:.4f}")
+             return margin
+        return None

src/strategies/momentum.py

@@ -0,0 +1,64 @@
+import logging
+from datetime import datetime, timezone
+import pandas as pd
+from typing import Optional, List
+
+logger = logging.getLogger(__name__)
+
+class NewsReactionMomentum:
+    def __init__(self, sentiment_threshold: float = 0.5, volume_spike_multiplier: float = 3.0):
+        """
+        Momentum strategy reacting to breaking news sentiment combined with volume confirmation.
+        """
+        self.sentiment_threshold = sentiment_threshold
+        self.volume_spike_multiplier = volume_spike_multiplier
+        
+        # Keep track of recent volume to calculate moving averages
+        self.volume_history = [] 
+        self.history_window = 60 # e.g., last 60 ticks/minutes
+
+    def update_volume(self, volume: float):
+        self.volume_history.append(volume)
+        if len(self.volume_history) > self.history_window:
+            self.volume_history.pop(0)
+
+    def _get_baseline_volume(self) -> float:
+        if not self.volume_history: return 0.0
+        return sum(self.volume_history) / len(self.volume_history)
+
+    def generate_signal(self, current_price: float, current_volume: float, sentiment_score: float) -> Optional[dict]:
+        """
+        Evaluate if current conditions warrant a momentum trade.
+        Requires both high sentiment magnitude AND a volume spike.
+        """
+        self.update_volume(current_volume)
+        
+        if len(self.volume_history) < 10:
+            return None # Not enough history for a baseline
+            
+        baseline_vol = self._get_baseline_volume()
+        is_volume_spike = current_volume > (baseline_vol * self.volume_spike_multiplier)
+        
+        if not is_volume_spike:
+            return None
+            
+        # Check if sentiment matches the directional threshold
+        if sentiment_score >= self.sentiment_threshold:
+            logger.info("Bullish momentum signal triggered (Positive News + Volume Spike)")
+            return {
+                "action": "BUY_YES",
+                "confidence": sentiment_score,
+                "target_price": min(1.0, current_price + 0.05), # Take profit +5 cents
+                "stop_loss": max(0.01, current_price - 0.02)    # Stop loss -2 cents
+            }
+            
+        elif sentiment_score <= -self.sentiment_threshold:
+            logger.info("Bearish momentum signal triggered (Negative News + Volume Spike)")
+            return {
+                "action": "SELL_YES", # Or BUY_NO depending on the market structure
+                "confidence": abs(sentiment_score),
+                "target_price": max(0.01, current_price - 0.05),
+                "stop_loss": min(1.0, current_price + 0.02)
+            }
+            
+        return None