def find_yield_opportunities(markets: list, min_prob: float = 0.95):
"""Find high-probability outcomes trading below $1."""
opportunities = []
for market in markets:
# Check if YES or NO has high probability
if market.yes_price >= min_prob:
outcome = "YES"
price = market.yes_price
elif market.no_price >= min_prob:
outcome = "NO"
price = market.no_price
else:
continue
# Calculate returns
profit_pct = ((1.0 - price) / price) * 100
days = days_until_resolution(market.end_date)
apy = (profit_pct / days) * 365 if days > 0 else 0
opportunities.append({
"market": market.question,
"outcome": outcome,
"price": price,
"profit": profit_pct,
"apy": apy
})
return sorted(opportunities, key=lambda x: x["apy"], reverse=True)
def find_threshold_mispricing(markets: list, topic: str = "BTC"):
"""Find mispriced threshold markets."""
# Filter to reach-type markets about the topic
reach_markets = [
m for m in markets
if topic in m.question.upper()
and any(word in m.question.lower() for word in ["reach", "hit", "above"])
]
# Extract threshold from each
for m in reach_markets:
m.threshold = extract_price_threshold(m.question)
# Sort by threshold
reach_markets.sort(key=lambda x: x.threshold)
# Check adjacent pairs
opportunities = []
for i in range(len(reach_markets) - 1):
lower = reach_markets[i] # e.g., BTC $100K
higher = reach_markets[i+1] # e.g., BTC $150K
# Mispricing if higher threshold has higher YES price
if higher.yes_price > lower.yes_price:
opportunities.append({
"error": f"${higher.threshold:,.0f} YES > ${lower.threshold:,.0f} YES",
"buy": f"${lower.threshold:,.0f} YES",
"expected_value": higher.yes_price,
"current_price": lower.yes_price,
"profit_potential": (higher.yes_price - lower.yes_price) / lower.yes_price * 100
})
return opportunities
def detect_momentum(market: Market, price_history: list) -> Signal:
"""Detect momentum signals in market."""
current = market.yes_price
price_24h_ago = get_price_at(price_history, hours=24)
if price_24h_ago is None:
return None
change = (current - price_24h_ago) / price_24h_ago
# Surge detection
if abs(change) > 0.10: # >10% move
return Signal(
type="SURGE",
direction="UP" if change > 0 else "DOWN",
magnitude=abs(change) * 100,
action="Investigate news, consider following trend"
)
return None
def recommend_strategy(market: Market, ai_consensus: Consensus) -> str:
"""Recommend best strategy for this market."""
# High probability + near-term = Yield Farming
if max(market.yes_price, market.no_price) > 0.90:
if market.days_to_resolution < 60:
return "YIELD_FARMING"
# Related markets exist = check Delta Neutral
related = find_related_markets(market)
if related:
inconsistencies = check_logical_consistency(market, related)
if inconsistencies:
return "DELTA_NEUTRAL"
# Recent price movement = Momentum
if abs(market.price_change_24h) > 0.05:
return "MOMENTUM"
# Default
return "STANDARD_ANALYSIS"
