File size: 3,795 Bytes
77fd2f6 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 | 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
|