Zion Boggan zionboggan.com ↗

initial commit

6678f98   Zion Boggan committed on Mar 30, 2026 (2 months ago)
eval/effective_exposure.py +79 -0
@@ -0,0 +1,79 @@
+"""
+Effective Exposure Discount Module for Hermes v4.3
+
+When bets are essentially won/lost (market price near $1.00 or $0.00) but Kalshi
+hasn't officially settled them yet, raw exposure stays locked and blocks new trades
+during prime hours. This module provides "effective exposure" that discounts
+near-certain outcomes so capital is freed for new opportunities.
+
+The actual settlement process is unchanged - this only affects the *trading gate*.
+"""
+
+DISCOUNT_TIERS = [
+ (0.95, 0.00),
+ (0.85, 0.25),
+ (0.75, 0.50),
+
+]
+
+def effective_exposure_factor(side, yes_price):
+ """
+ Given a trade's side ('yes' or 'no') and the current YES price on Kalshi,
+ return the fraction of cost that should count as effective exposure.
+
+ Returns float in [0.0, 1.0].
+ """
+ if yes_price is None:
+ return 1.0
+
+ if side == "yes":
+ win_price = yes_price
+ else:
+ win_price = 1.0 - yes_price
+
+ for threshold, factor in DISCOUNT_TIERS:
+ if win_price >= threshold:
+ return factor
+
+ return 1.0
+
+def compute_effective_exposure(open_trades, price_lookup):
+ """
+ Compute total effective exposure across all open trades.
+
+ Args:
+ open_trades: list of dicts with keys: id, ticker, side, cost
+ price_lookup: dict mapping ticker -> current YES price (float)
+ or a callable(ticker) -> float
+
+ Returns:
+ (effective_total, details) where details is a list of per-trade info dicts
+ """
+ effective_total = 0.0
+ details = []
+
+ for trade in open_trades:
+ ticker = trade["ticker"]
+ side = trade["side"]
+ cost = float(trade["cost"])
+
+ if callable(price_lookup):
+ yes_price = price_lookup(ticker)
+ else:
+ yes_price = price_lookup.get(ticker)
+
+ factor = effective_exposure_factor(side, yes_price)
+ effective_cost = round(cost * factor, 2)
+ effective_total += effective_cost
+
+ details.append({
+ "id": trade.get("id"),
+ "ticker": ticker,
+ "side": side,
+ "cost": cost,
+ "yes_price": yes_price,
+ "factor": factor,
+ "effective_cost": effective_cost,
+ })
+
+ return round(effective_total, 2), details
eval/test_effective_exposure.py +481 -0
@@ -0,0 +1,481 @@
+"""
+Simulation Test Suite for Effective Exposure Discount
+=====================================================
+
+Tests the proposed change in isolation WITHOUT touching production code.
+Mocks the database and Kalshi API to simulate real-world scenarios.
+
+Run: python test_effective_exposure.py
+"""
+
+import sys
+import unittest
+from unittest.mock import patch, MagicMock
+from datetime import datetime, timedelta, timezone
+
+from effective_exposure import (
+ effective_exposure_factor,
+ compute_effective_exposure,
+ DISCOUNT_TIERS,
+)
+
+MAX_EXPOSURE_PCT = 0.30
+MAX_BET_PCT = 0.08
+
+def old_exposure_gate(open_trades, balance):
+ """Original gate: raw SUM(cost) of open trades."""
+ raw_exposure = sum(float(t["cost"]) for t in open_trades)
+ limit = balance * MAX_EXPOSURE_PCT
+ return raw_exposure < limit, raw_exposure, limit
+
+def new_exposure_gate(open_trades, price_lookup, balance):
+ """New gate: effective exposure with discounts."""
+ effective, details = compute_effective_exposure(open_trades, price_lookup)
+ limit = balance * MAX_EXPOSURE_PCT
+ return effective < limit, effective, limit, details
+
+class TestExposureFactor(unittest.TestCase):
+ """Verify discount tiers work correctly for YES and NO sides."""
+
+ def test_yes_side_essentially_won(self):
+ """YES bet, market price 0.97 → 0% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.97), 0.00)
+
+ def test_yes_side_very_likely(self):
+ """YES bet, market price 0.88 → 25% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.88), 0.25)
+
+ def test_yes_side_probably_winning(self):
+ """YES bet, market price 0.78 → 50% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.78), 0.50)
+
+ def test_yes_side_uncertain(self):
+ """YES bet, market price 0.60 → full exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.60), 1.0)
+
+ def test_yes_side_losing(self):
+ """YES bet, market price 0.15 → full exposure (losing)."""
+ self.assertEqual(effective_exposure_factor("yes", 0.15), 1.0)
+
+ def test_no_side_essentially_won(self):
+ """NO bet, YES price 0.03 → NO win_price=0.97 → 0% exposure."""
+ self.assertEqual(effective_exposure_factor("no", 0.03), 0.00)
+
+ def test_no_side_very_likely(self):
+ """NO bet, YES price 0.12 → NO win_price=0.88 → 25% exposure."""
+ self.assertEqual(effective_exposure_factor("no", 0.12), 0.25)
+
+ def test_no_side_uncertain(self):
+ """NO bet, YES price 0.55 → NO win_price=0.45 → full exposure."""
+ self.assertEqual(effective_exposure_factor("no", 0.55), 1.0)
+
+ def test_no_side_losing(self):
+ """NO bet, YES price 0.90 → NO win_price=0.10 → full exposure."""
+ self.assertEqual(effective_exposure_factor("no", 0.90), 1.0)
+
+ def test_none_price_conservative(self):
+ """No price data → always full exposure."""
+ self.assertEqual(effective_exposure_factor("yes", None), 1.0)
+ self.assertEqual(effective_exposure_factor("no", None), 1.0)
+
+ def test_exact_boundary_95(self):
+ """Exactly at 0.95 threshold → 0% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.95), 0.00)
+
+ def test_just_below_95(self):
+ """Just below 0.95 → falls to 25% tier."""
+ self.assertEqual(effective_exposure_factor("yes", 0.9499), 0.25)
+
+ def test_exact_boundary_85(self):
+ """Exactly at 0.85 threshold → 25% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.85), 0.25)
+
+ def test_exact_boundary_75(self):
+ """Exactly at 0.75 threshold → 50% exposure."""
+ self.assertEqual(effective_exposure_factor("yes", 0.75), 0.50)
+
+class TestComputeEffective(unittest.TestCase):
+
+ def test_all_uncertain_matches_raw(self):
+ """When all positions are uncertain, effective == raw exposure."""
+ trades = [
+ {"id": 1, "ticker": "KXHIGHNY-26MAR31-B62.5", "side": "yes", "cost": 5.00},
+ {"id": 2, "ticker": "KXHIGHCH-26MAR31-B55.5", "side": "no", "cost": 3.00},
+ ]
+ prices = {
+ "KXHIGHNY-26MAR31-B62.5": 0.60,
+ "KXHIGHCH-26MAR31-B55.5": 0.50,
+ }
+ effective, details = compute_effective_exposure(trades, prices)
+ raw = sum(t["cost"] for t in trades)
+ self.assertEqual(effective, raw)
+
+ def test_one_won_one_uncertain(self):
+ """One essentially won + one uncertain → only uncertain counts."""
+ trades = [
+ {"id": 1, "ticker": "WON-TICKER", "side": "yes", "cost": 5.00},
+ {"id": 2, "ticker": "OPEN-TICKER", "side": "yes", "cost": 3.00},
+ ]
+ prices = {
+ "WON-TICKER": 0.97,
+ "OPEN-TICKER": 0.55,
+ }
+ effective, details = compute_effective_exposure(trades, prices)
+ self.assertEqual(effective, 3.00)
+
+ def test_mixed_discounts(self):
+ """Multiple positions at different tiers."""
+ trades = [
+ {"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00},
+ {"id": 2, "ticker": "T2", "side": "yes", "cost": 8.00},
+ {"id": 3, "ticker": "T3", "side": "no", "cost": 6.00},
+ {"id": 4, "ticker": "T4", "side": "yes", "cost": 4.00},
+ ]
+ prices = {
+ "T1": 0.96,
+ "T2": 0.88,
+ "T3": 0.22,
+ "T4": 0.60,
+ }
+ effective, details = compute_effective_exposure(trades, prices)
+ expected = 0.00 + 2.00 + 3.00 + 4.00
+ self.assertAlmostEqual(effective, expected, places=2)
+
+ def test_callable_price_lookup(self):
+ """Price lookup via callable (simulating API call)."""
+ trades = [
+ {"id": 1, "ticker": "T1", "side": "yes", "cost": 5.00},
+ ]
+
+ def lookup(ticker):
+ return 0.97 if ticker == "T1" else 0.50
+
+ effective, details = compute_effective_exposure(trades, lookup)
+ self.assertEqual(effective, 0.00)
+
+ def test_missing_price_in_dict(self):
+ """Ticker not in price dict → full exposure (conservative)."""
+ trades = [
+ {"id": 1, "ticker": "UNKNOWN", "side": "yes", "cost": 5.00},
+ ]
+ effective, details = compute_effective_exposure(trades, {})
+ self.assertEqual(effective, 5.00)
+
+ def test_empty_trades(self):
+ """No open trades → zero exposure."""
+ effective, details = compute_effective_exposure([], {})
+ self.assertEqual(effective, 0.00)
+ self.assertEqual(details, [])
+
+class TestOvernightLockupScenario(unittest.TestCase):
+ """
+ Scenario: It's 8 PM. Hermes placed 3 weather bets during the afternoon
+ scanning cycle. Total cost = $25 on a $100 bankroll (25% exposure).
+
+ By 6 AM the next morning, 2 of the 3 bets are essentially decided
+ (market prices at 0.96 and 0.98) but Kalshi won't settle until 10 AM.
+
+ OLD behavior: $25 exposure stays locked → only $5 room before 30% cap.
+ NEW behavior: Only the uncertain bet's cost counts → much more room.
+ """
+
+ def setUp(self):
+ self.balance = 100.00
+ self.open_trades = [
+ {"id": 1, "ticker": "KXHIGHNY-26MAR30-B62.5", "side": "yes", "cost": 10.00},
+ {"id": 2, "ticker": "KXHIGHCH-26MAR30-B50.5", "side": "yes", "cost": 8.00},
+ {"id": 3, "ticker": "KXHIGHDN-26MAR30-B55.5", "side": "no", "cost": 7.00},
+ ]
+
+ def test_old_gate_blocks_morning_trading(self):
+ """OLD system: $25 exposure, only $5 room → a $6 bet is BLOCKED."""
+ can_trade, raw, limit = old_exposure_gate(self.open_trades, self.balance)
+
+ self.assertTrue(can_trade)
+ self.assertEqual(raw, 25.00)
+ self.assertEqual(limit, 30.00)
+
+ room = limit - raw
+ self.assertEqual(room, 5.00)
+ self.assertFalse(room >= 6.00, "Old gate: no room for a $6 bet")
+
+ def test_new_gate_frees_morning_capital(self):
+ """
+ NEW system: 2 bets essentially won, 1 uncertain.
+ NY at 0.96 (won) → $0 effective
+ CH at 0.98 (won) → $0 effective
+ DN NO side, YES price 0.20NO win_price=0.8050% tier → $3.50
+ Total effective = $3.50, room = $26.50
+ """
+ prices = {
+ "KXHIGHNY-26MAR30-B62.5": 0.96,
+ "KXHIGHCH-26MAR30-B50.5": 0.98,
+ "KXHIGHDN-26MAR30-B55.5": 0.20,
+ }
+ can_trade, effective, limit, details = new_exposure_gate(
+ self.open_trades, prices, self.balance
+ )
+ self.assertTrue(can_trade)
+ self.assertAlmostEqual(effective, 3.50, places=2)
+ room = limit - effective
+ self.assertAlmostEqual(room, 26.50, places=2)
+ self.assertTrue(room >= 6.00, "New gate: plenty of room for morning bets!")
+
+ def test_improvement_quantified(self):
+ """Quantify: new system gives 5.3x more trading room in this scenario."""
+ prices = {
+ "KXHIGHNY-26MAR30-B62.5": 0.96,
+ "KXHIGHCH-26MAR30-B50.5": 0.98,
+ "KXHIGHDN-26MAR30-B55.5": 0.20,
+ }
+ _, raw, limit = old_exposure_gate(self.open_trades, self.balance)
+ _, effective, _, _ = new_exposure_gate(self.open_trades, prices, self.balance)
+
+ old_room = limit - raw
+ new_room = limit - effective
+ improvement = new_room / old_room
+ self.assertGreater(improvement, 5.0)
+ print(f"\n >>> Overnight scenario: old room=${old_room:.2f}, "
+ f"new room=${new_room:.2f} ({improvement:.1f}x improvement)")
+
+class TestSafetyFlipScenario(unittest.TestCase):
+ """
+ Edge case: A bet LOOKED like it was winning (price at 0.90) but then
+ the market reverses. If we discounted it, are we over-exposed?
+
+ The 0.75 threshold with 50% discount is intentionally conservative.
+ A bet at 0.90 gets 25% discount - still counts 75% of its cost.
+ Only bets at 0.95+ get fully discounted.
+ """
+
+ def test_moderate_winner_still_counts(self):
+ """Bet at 0.80 YES → 50% discount, still counts half."""
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00}]
+
+ effective, _ = compute_effective_exposure(trades, {"T1": 0.80})
+ self.assertEqual(effective, 5.00)
+
+ def test_strong_winner_minimal_exposure(self):
+ """Bet at 0.90 YES → 25% discount, counts quarter."""
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00}]
+ effective, _ = compute_effective_exposure(trades, {"T1": 0.90})
+ self.assertEqual(effective, 2.50)
+
+ def test_near_certain_zero_exposure(self):
+ """Only at 0.95+ does exposure drop to zero."""
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00}]
+ effective, _ = compute_effective_exposure(trades, {"T1": 0.95})
+ self.assertEqual(effective, 0.00)
+
+ def test_worst_case_total_reversal(self):
+ """
+ Worst case: we freed exposure on a 0.95 bet, placed a new trade,
+ then the 0.95 bet crashes to 0.30. Now we have MORE real exposure
+ than the 30% cap intended.
+
+ BUT: On Kalshi weather markets, a price at 0.95 means the weather
+ event is almost certainly decided. These are binary temperature
+ outcomes - they don't "reverse" the way stock prices can.
+
+ Still, let's quantify the max theoretical over-exposure.
+ """
+ balance = 100.00
+
+ original_trades = [
+ {"id": 1, "ticker": "WON", "side": "yes", "cost": 20.00},
+ {"id": 2, "ticker": "OPEN", "side": "yes", "cost": 8.00},
+ ]
+
+ prices_before = {"WON": 0.96, "OPEN": 0.55}
+ _, effective_before, _, _ = new_exposure_gate(original_trades, prices_before, balance)
+ self.assertAlmostEqual(effective_before, 8.00, places=2)
+
+ trades_after = original_trades + [
+ {"id": 3, "ticker": "NEW", "side": "yes", "cost": 8.00},
+ ]
+
+ prices_after = {"WON": 0.30, "OPEN": 0.55, "NEW": 0.55}
+ _, effective_after, _, _ = new_exposure_gate(trades_after, prices_after, balance)
+
+ self.assertAlmostEqual(effective_after, 36.00, places=2)
+
+ overshoot_pct = (effective_after / balance) - MAX_EXPOSURE_PCT
+ self.assertLessEqual(overshoot_pct, MAX_BET_PCT,
+ "Overshoot is bounded by MAX_BET_PCT")
+ print(f"\n >>> Worst-case reversal: {effective_after/balance:.0%} effective "
+ f"exposure (overshoot={overshoot_pct:.0%}, bounded by MAX_BET_PCT={MAX_BET_PCT:.0%})")
+
+class TestFullPipelineSimulation(unittest.TestCase):
+ """
+ Simulates:
+ 1. Afternoon: Bot places 3 bets, hitting 24% raw exposure
+ 2. Evening: Markets move, 2 bets are near-certain wins
+ 3. Overnight: No settlement from Kalshi
+ 4. Morning: New scanning cycle - can the bot trade?
+ 5. Mid-morning: Kalshi settles, exposure clears naturally
+ """
+
+ def test_full_24_hour_cycle(self):
+ balance = 100.00
+ events = []
+
+ trades = [
+ {"id": 1, "ticker": "NY-HIGH", "side": "yes", "cost": 9.00},
+ {"id": 2, "ticker": "CH-HIGH", "side": "yes", "cost": 8.00},
+ {"id": 3, "ticker": "DN-LOW", "side": "no", "cost": 7.00},
+ ]
+ prices_afternoon = {"NY-HIGH": 0.58, "CH-HIGH": 0.62, "DN-LOW": 0.45}
+
+ _, eff_afternoon, _, _ = new_exposure_gate(trades, prices_afternoon, balance)
+
+ self.assertAlmostEqual(eff_afternoon, 24.00, places=2)
+ events.append(f"3 PM: Placed 3 bets, effective exposure=${eff_afternoon:.2f}")
+
+ prices_evening = {"NY-HIGH": 0.92, "CH-HIGH": 0.88, "DN-LOW": 0.10}
+ _, eff_evening, _, _ = new_exposure_gate(trades, prices_evening, balance)
+
+ self.assertAlmostEqual(eff_evening, 6.00, places=2)
+ events.append(f"9 PM: Markets shifted, effective exposure=${eff_evening:.2f}")
+
+ prices_overnight = {"NY-HIGH": 0.97, "CH-HIGH": 0.96, "DN-LOW": 0.04}
+ _, eff_overnight, _, _ = new_exposure_gate(trades, prices_overnight, balance)
+
+ self.assertAlmostEqual(eff_overnight, 0.00, places=2)
+ events.append(f"2 AM: Near-certain, effective exposure=${eff_overnight:.2f}")
+
+ room = (balance * MAX_EXPOSURE_PCT) - eff_overnight
+ self.assertAlmostEqual(room, 30.00, places=2)
+
+ old_room = (balance * MAX_EXPOSURE_PCT) - 24.00
+ self.assertEqual(old_room, 6.00)
+
+ events.append(f"6 AM: OLD room=${old_room:.2f}, NEW room=${room:.2f}")
+ events.append(f" Improvement: {room/old_room:.1f}x more capital available")
+
+ morning_bet_cost = 7.50
+ trades.append({"id": 4, "ticker": "LA-HIGH", "side": "yes", "cost": morning_bet_cost})
+ prices_morning = {**prices_overnight, "LA-HIGH": 0.55}
+ _, eff_morning, _, _ = new_exposure_gate(trades, prices_morning, balance)
+ self.assertAlmostEqual(eff_morning, 7.50, places=2)
+ events.append(f"6:30 AM: Placed morning bet, effective=${eff_morning:.2f}")
+
+ trades_after_settle = [trades[3]]
+ _, eff_settled, _, _ = new_exposure_gate(trades_after_settle, {"LA-HIGH": 0.55}, balance)
+ self.assertAlmostEqual(eff_settled, 7.50, places=2)
+ events.append(f"10 AM: Kalshi settled 3 bets, exposure=${eff_settled:.2f}")
+
+ print("\n >>> Full 24-hour simulation:")
+ for e in events:
+ print(f" {e}")
+
+class TestIntegrationCompatibility(unittest.TestCase):
+ """
+ Verify the new function can serve as a drop-in replacement for
+ get_open_exposure() in the Filter 5 check at line 2108-2115.
+ """
+
+ def test_when_no_prices_matches_old_behavior(self):
+ """Without price data, effective == raw (backward compatible)."""
+ trades = [
+ {"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00},
+ {"id": 2, "ticker": "T2", "side": "no", "cost": 5.00},
+ ]
+
+ effective, _ = compute_effective_exposure(trades, {})
+ raw = sum(t["cost"] for t in trades)
+ self.assertEqual(effective, raw)
+
+ def test_gate_decision_matches_types(self):
+ """The gate returns a bool just like the old comparison."""
+ balance = 100.0
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 25.00}]
+ prices = {"T1": 0.97}
+
+ can_trade, effective, limit, _ = new_exposure_gate(trades, prices, balance)
+ self.assertIsInstance(can_trade, bool)
+ self.assertIsInstance(effective, float)
+ self.assertIsInstance(limit, float)
+
+ def test_does_not_modify_input(self):
+ """Ensure the function doesn't mutate the input trade list."""
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 10.00}]
+ original = [dict(t) for t in trades]
+ compute_effective_exposure(trades, {"T1": 0.97})
+ self.assertEqual(trades, original)
+
+class TestStress(unittest.TestCase):
+
+ def test_max_positions_at_various_states(self):
+ """8 trades (MAX_DAILY_TRADES), all at different price levels."""
+ trades = [
+ {"id": i, "ticker": f"T{i}", "side": "yes", "cost": 4.00}
+ for i in range(1, 9)
+ ]
+
+ prices = {
+ "T1": 0.99,
+ "T2": 0.97,
+ "T3": 0.95,
+ "T4": 0.90,
+ "T5": 0.80,
+ "T6": 0.60,
+ "T7": 0.45,
+ "T8": 0.10,
+ }
+ effective, details = compute_effective_exposure(trades, prices)
+
+ self.assertAlmostEqual(effective, 15.00, places=2)
+
+ raw = sum(t["cost"] for t in trades)
+ reduction = 1 - (effective / raw)
+ print(f"\n >>> Stress test: 8 positions, raw=${raw:.2f}, "
+ f"effective=${effective:.2f} ({reduction:.0%} reduction)")
+
+ def test_all_won_zero_exposure(self):
+ """All positions essentially won → zero exposure."""
+ trades = [
+ {"id": i, "ticker": f"T{i}", "side": "yes", "cost": 5.00}
+ for i in range(1, 5)
+ ]
+ prices = {f"T{i}": 0.98 for i in range(1, 5)}
+ effective, _ = compute_effective_exposure(trades, prices)
+ self.assertEqual(effective, 0.00)
+
+ def test_all_lost_full_exposure(self):
+ """All positions are losing → still full exposure (conservative)."""
+ trades = [
+ {"id": i, "ticker": f"T{i}", "side": "yes", "cost": 5.00}
+ for i in range(1, 5)
+ ]
+ prices = {f"T{i}": 0.10 for i in range(1, 5)}
+ effective, _ = compute_effective_exposure(trades, prices)
+ self.assertEqual(effective, 20.00)
+
+class TestDetailsOutput(unittest.TestCase):
+ """The details list should be usable for Discord reporting."""
+
+ def test_details_contain_all_fields(self):
+ trades = [{"id": 1, "ticker": "T1", "side": "yes", "cost": 5.00}]
+ _, details = compute_effective_exposure(trades, {"T1": 0.90})
+ d = details[0]
+ self.assertIn("id", d)
+ self.assertIn("ticker", d)
+ self.assertIn("side", d)
+ self.assertIn("cost", d)
+ self.assertIn("yes_price", d)
+ self.assertIn("factor", d)
+ self.assertIn("effective_cost", d)
+
+ def test_details_count_matches_trades(self):
+ trades = [
+ {"id": i, "ticker": f"T{i}", "side": "yes", "cost": 3.00}
+ for i in range(1, 4)
+ ]
+ _, details = compute_effective_exposure(trades, {})
+ self.assertEqual(len(details), 3)
+
+if __name__ == "__main__":
+ print("=" * 70)
+ print("HERMES EFFECTIVE EXPOSURE DISCOUNT - SIMULATION TEST SUITE")
+ print("=" * 70)
+ unittest.main(verbosity=2)