forked from kevinrgu/autoagent
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtrade.py
More file actions
7401 lines (6575 loc) · 356 KB
/
trade.py
File metadata and controls
7401 lines (6575 loc) · 356 KB
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
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
import os, time, base64, json, warnings, sys, sqlite3, urllib.parse, ssl, uuid
warnings.filterwarnings("ignore", category=DeprecationWarning) # only suppress deprecation noise
from datetime import datetime, timezone, timedelta
# Fix SSL cert verification on macOS (Python may not find system certs)
# truststore hooks into macOS Keychain for full cert coverage
try:
import truststore
truststore.inject_into_ssl()
except ImportError:
try:
import certifi
os.environ.setdefault("SSL_CERT_FILE", certifi.where())
os.environ.setdefault("REQUESTS_CA_BUNDLE", certifi.where())
except ImportError:
_mac_cert = "/etc/ssl/cert.pem"
if os.path.exists(_mac_cert):
os.environ.setdefault("SSL_CERT_FILE", _mac_cert)
os.environ.setdefault("REQUESTS_CA_BUNDLE", _mac_cert)
try:
import requests
except ImportError:
os.system(f"{sys.executable} -m pip install requests cryptography certifi -q")
import requests
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import padding
# New modular sources — import from bot package if available
try:
from bot.signals.sources.metar_observations import (
get_metar_observation_estimate,
is_metar_fresh_for_ticker as _is_metar_fresh_for_ticker,
)
except ImportError:
get_metar_observation_estimate = None
_is_metar_fresh_for_ticker = None
try:
from bot.signals.sources.fedwatch import get_fedwatch_estimate
except ImportError:
get_fedwatch_estimate = None
try:
from bot.signals.sources.deribit_vol import get_deribit_term_vol, get_deribit_implied_prob
except ImportError:
get_deribit_term_vol = None
get_deribit_implied_prob = None
# ══════════════════════════════════════════════════════════════════════════════
# CONFIG
# ══════════════════════════════════════════════════════════════════════════════
KEY_ID = os.environ.get("KALSHI_API_KEY_ID", "")
KEY_PATH = os.environ.get("KALSHI_PRIVATE_KEY_PATH", "")
if not KEY_ID or not KEY_PATH:
print("[FATAL] KALSHI_API_KEY_ID and KALSHI_PRIVATE_KEY_PATH must be set in environment")
# Don't exit — allow import for debugging, but trading will fail
BASE_URL = os.environ.get("KALSHI_API_BASE", "https://api.elections.kalshi.com/trade-api/v2")
HOST = BASE_URL.split("/trade-api")[0]
DRY_RUN = os.environ.get("DRY_RUN", "false").lower() in ("true", "1", "yes")
DAILY_LOSS_LIMIT = float(os.environ.get("DAILY_LOSS_LIMIT_PCT", "0.10"))
MAX_DRAWDOWN = float(os.environ.get("MAX_DRAWDOWN_PCT", "0.15"))
KELLY_FRACTION = float(os.environ.get("KELLY_FRACTION", "0.10"))
MAX_CONTRACTS = int(os.environ.get("MAX_CONTRACTS", "500")) # hard ceiling (dynamic sizing via MAX_POSITION_PCT is primary control)
MAX_POSITION_PCT = float(os.environ.get("MAX_POSITION_PCT", "0.02")) # max 2% of balance per position
DB_PATH = os.environ.get("DB_PATH", "/task/kalshi_trades.db")
MIN_WIN_RATE = float(os.environ.get("MIN_WIN_RATE", "0.45"))
MIN_SAMPLE_SIZE = int(os.environ.get("MIN_SAMPLE_SIZE", "5"))
ORDER_MAX_AGE_HOURS = float(os.environ.get("ORDER_MAX_AGE_HOURS", "2"))
# Position management
TAKE_PROFIT_PCT = float(os.environ.get("TAKE_PROFIT_PCT", "0.20")) # sell at +20%
STOP_LOSS_PCT = float(os.environ.get("STOP_LOSS_PCT", "0.15")) # sell at -15%
MAX_HOLD_DAYS = int(os.environ.get("MAX_HOLD_DAYS", "7")) # force exit after 7d
# Information layer
MIN_EDGE = float(os.environ.get("MIN_EDGE", "0.07")) # require 7% edge over market (3+ sources)
SINGLE_SOURCE_EDGE = float(os.environ.get("SINGLE_SOURCE_EDGE", "0.12")) # 12% edge with only 1 source
MAX_PER_CATEGORY = int(os.environ.get("MAX_PER_CATEGORY", "2")) # max concurrent positions per risk category
MAX_PORTFOLIO_PCT = float(os.environ.get("MAX_PORTFOLIO_PCT", "0.15")) # max 15% of balance in open positions
# ── Immutable initial-config snapshot ─────────────────────────────────────────
# Captured once at module import. apply_phase_limits() and the active-feedback
# block MUST derive effective values from these, not from the current mutated
# globals, otherwise each daemon cycle compounds the multiplication and drives
# KELLY→0 / MIN_EDGE→∞ as process lifetime grows. See tests/test_config_no_drift.py.
_INITIAL_DRY_RUN = DRY_RUN
_INITIAL_MAX_POSITION_PCT = MAX_POSITION_PCT
_INITIAL_MAX_PORTFOLIO_PCT = MAX_PORTFOLIO_PCT
_INITIAL_MAX_CONTRACTS = MAX_CONTRACTS
_INITIAL_KELLY_FRACTION = KELLY_FRACTION
_INITIAL_MIN_EDGE = MIN_EDGE
_INITIAL_SINGLE_SOURCE_EDGE = SINGLE_SOURCE_EDGE
# Fee accounting: use exact Kalshi formulas from bot/core/money.py
# Old flat estimate ESTIMATED_FEE_PER_CONTRACT=0.03 was ~3-7x too high for maker orders.
# Now we compute price-dependent fees: maker entry + taker exit per contract.
from bot.core.money import fee_per_contract_cents as _fee_per_contract_cents
from bot.learning.alpha_log import (
DecisionOutcome as _AlphaOutcome,
DecisionType as _AlphaType,
EnsembleSnapshot as _AlphaEnsemble,
fill_settlement_for_ticker as _alpha_fill_settlement,
log_decision as _alpha_log_decision,
market_snapshot_from_dict as _alpha_market_snapshot,
to_canonical_p_yes as _to_canonical_p_yes,
)
from bot.learning.populate_from_alpha import populate_all as _alpha_populate_all
from bot.learning.directional_shadow import (
ShadowOutcome as _ShadowOutcome,
evaluate as _eval_directional_shadow,
get_kelly_multiplier as _get_kelly_multiplier,
get_live_state as _get_live_state,
LiveState as _LiveState,
)
from bot.learning.alpha_log import family_from_ticker as _family_from_ticker
from bot.learning.calibration import (
apply_calibration as _apply_calibration,
apply_calibration_correction,
compute_calibration_correction,
fit_and_persist as _cal_fit_and_persist,
load_curve as _cal_load_curve,
reset_cache as _cal_reset_cache,
)
ESTIMATED_EXIT_SPREAD = float(os.environ.get("ESTIMATED_EXIT_SPREAD", "0.03")) # 3¢ expected exit slippage
def _round_trip_fee_dollars(price_dollars: float) -> float:
"""Round-trip fee per contract in dollars (maker entry + taker exit).
At 50¢: ~2.2¢ (was 6¢ flat — 3x too high)
At 20¢: ~1.4¢ (was 6¢ flat — 4x too high)
At 80¢: ~1.4¢ (was 6¢ flat — 4x too high)
"""
pc = max(1, min(99, round(price_dollars * 100)))
entry = _fee_per_contract_cents(pc, maker=True) # ~0.44¢ at 50¢
exit_ = _fee_per_contract_cents(pc, maker=False) # ~1.75¢ at 50¢
return (entry + exit_) / 100 # convert cents to dollars
def log_opportunity(conn, ticker, strategy, action, side=None, ensemble_prob=None,
market_prob=None, edge=None, source_count=None,
sources_json=None, skip_reason=None):
"""Log a market candidate to the opportunity_log decision journal.
Called for every candidate — traded or rejected — so we can analyze
'what did we miss?' and 'why did we take that?' after the fact.
"""
try:
conn.execute("""INSERT INTO opportunity_log
(ticker, strategy, action, side, ensemble_prob, market_prob,
edge, source_count, sources_json, skip_reason)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""",
(ticker, strategy, action, side, ensemble_prob, market_prob,
edge, source_count, sources_json, skip_reason))
except Exception:
pass # Never let logging break the trading loop
# Market making config was deleted in Phase 0 of the MM-deletion pivot (2026-04-16).
# Negative P&L after 813 fills + 69 losing settlements forced the decision to remove
# the entire market-maker package. See /Users/jlu/.claude/plans/elegant-petting-dawn.md
# for the roadmap. If we ever want MM back it will be a new, from-scratch project.
# Safe Compounder config (separate from directional)
SC_ENABLED = os.environ.get("SC_ENABLED", "true").lower() in ("true", "1", "yes")
SC_DRY_RUN = os.environ.get("SC_DRY_RUN", "true").lower() in ("true", "1", "yes")
# ══════════════════════════════════════════════════════════════════════════════
# PHASED SIZING — auto-ramp based on track record
# ══════════════════════════════════════════════════════════════════════════════
# The bot starts ultra-conservative and only scales up as it earns the right to.
# Phase transitions are based on settled trade count AND win rate — both must be met.
# Override via FORCE_PHASE env var to manually lock a phase (e.g., "1" for paper only).
FORCE_PHASE = os.environ.get("FORCE_PHASE", "")
PHASE_CONFIG = {
# phase: (min_settled, min_win_rate, max_position_pct, max_portfolio_pct,
# max_contracts, kelly_mult, min_edge_mult, description)
1: (0, 0.00, 0.000, 0.000, 0, 0.0, 1.0,
"Paper trading — DRY_RUN forced on, zero risk"),
2: (50, 0.50, 0.035, 0.15, 10, 0.25, 1.5,
"Micro live — ~$20 max position, 10 contracts max, 1.5x edge required"),
3: (150, 0.52, 0.050, 0.05, 50, 0.50, 1.25,
"Small live — $500 max position, $5k max portfolio, 1.25x edge required"),
4: (300, 0.53, 0.010, 0.10, 200, 0.75, 1.1,
"Medium live — $1k max position, $10k max portfolio"),
5: (500, 0.54, 0.020, 0.15, 500, 1.00, 1.0,
"Full deployment — standard parameters"),
}
def compute_current_phase(conn):
"""Determine which phase the bot has earned based on its track record.
Returns (phase_number, phase_config_dict, stats_dict)."""
# Count settled trades and win rate
try:
row = conn.execute(
"SELECT COUNT(*), COALESCE(SUM(won), 0) FROM settlements"
).fetchone()
n_settled = row[0] if row else 0
n_won = row[1] if row else 0
win_rate = n_won / n_settled if n_settled > 0 else 0.0
# Also compute recent win rate (last 100 trades) for regression detection
recent = conn.execute(
"SELECT COUNT(*), COALESCE(SUM(won), 0) FROM "
"(SELECT won FROM settlements ORDER BY id DESC LIMIT 100)"
).fetchone()
recent_n = recent[0] if recent else 0
recent_wr = recent[1] / recent_n if recent_n > 0 else 0.0
except Exception:
n_settled, n_won, win_rate, recent_n, recent_wr = 0, 0, 0.0, 0, 0.0
stats = {
"settled": n_settled, "won": n_won, "win_rate": win_rate,
"recent_n": recent_n, "recent_win_rate": recent_wr,
}
# Manual override
if FORCE_PHASE:
try:
forced = int(FORCE_PHASE)
if forced in PHASE_CONFIG:
print(f"[phase] FORCED to Phase {forced} via env var")
return forced, PHASE_CONFIG[forced], stats
except ValueError:
pass
# Determine highest phase we qualify for
current_phase = 1
for phase_num in sorted(PHASE_CONFIG.keys()):
min_trades, min_wr, *_ = PHASE_CONFIG[phase_num]
if n_settled >= min_trades and (win_rate >= min_wr or n_settled == 0):
current_phase = phase_num
else:
break # phases are sequential — stop at first unqualified
# Safety: if recent win rate (last 100) drops below 48%, downgrade one phase
if recent_n >= 30 and recent_wr < 0.48 and current_phase > 2:
print(f"[phase] WARNING: Recent win rate {recent_wr:.1%} < 48% — "
f"downgrading from Phase {current_phase} to {current_phase - 1}")
current_phase -= 1
return current_phase, PHASE_CONFIG[current_phase], stats
def apply_phase_limits(phase_num, phase_cfg):
"""Derive phase-adjusted trading limits from the immutable INITIAL snapshot.
This function is idempotent: calling it N times in a long-lived daemon with
the same inputs always produces the same output. It never reads the current
mutated globals, so per-cycle multiplications cannot compound over the
process lifetime. Globals are still *written* (for read-compat with the
rest of trade.py) but always from _INITIAL_* values.
Returns a dict of the effective limits for logging and downstream threading.
"""
global DRY_RUN, MAX_POSITION_PCT, MAX_PORTFOLIO_PCT, MAX_CONTRACTS
global KELLY_FRACTION, MIN_EDGE, SINGLE_SOURCE_EDGE
_, _, max_pos_pct, max_port_pct, max_contracts, kelly_mult, edge_mult, desc = phase_cfg
# DIRECTIONAL TRADING DISABLED (V4): losing -$93.93 at 16% win rate.
DRY_RUN = True
# Take the more conservative of (user-configured initial, phase cap).
MAX_POSITION_PCT = min(_INITIAL_MAX_POSITION_PCT, max_pos_pct)
MAX_PORTFOLIO_PCT = min(_INITIAL_MAX_PORTFOLIO_PCT, max_port_pct)
MAX_CONTRACTS = min(_INITIAL_MAX_CONTRACTS, max_contracts)
# Multiplicative params derive from INITIAL, not current globals.
KELLY_FRACTION = _INITIAL_KELLY_FRACTION * kelly_mult
MIN_EDGE = _INITIAL_MIN_EDGE * edge_mult
SINGLE_SOURCE_EDGE = _INITIAL_SINGLE_SOURCE_EDGE * edge_mult
effective = {
"phase": phase_num, "description": desc,
"dry_run": DRY_RUN, "max_position_pct": MAX_POSITION_PCT,
"max_portfolio_pct": MAX_PORTFOLIO_PCT, "max_contracts": MAX_CONTRACTS,
"kelly_fraction": KELLY_FRACTION, "min_edge": MIN_EDGE,
"single_source_edge": SINGLE_SOURCE_EDGE,
}
return effective
# compute_dynamic_sizing: canonical implementation in bot/config.py
from bot.config import compute_dynamic_sizing # noqa: E402
print(f"[trade.py] HOST={HOST} DRY_RUN={DRY_RUN} "
f"KELLY={KELLY_FRACTION}x MIN_EDGE={MIN_EDGE} SINGLE_SRC_EDGE={SINGLE_SOURCE_EDGE}")
# ══════════════════════════════════════════════════════════════════════════════
# SQLite
# ══════════════════════════════════════════════════════════════════════════════
def init_db():
"""Initialize database — delegates to bot.db (canonical schema authority).
Returns sqlite3.Connection with all tables and migrations applied.
Also runs startup invariants to catch schema drift.
"""
from bot.db import init_db as _canonical_init_db, check_startup_invariants
conn = _canonical_init_db()
# ── Startup invariant check ──
failures = check_startup_invariants(conn)
if failures:
print("[init_db] ⚠️ STARTUP INVARIANT FAILURES:")
for f in failures:
print(f" ✗ {f}")
print("[init_db] The bot will continue but some features may be degraded.")
return conn
# ══════════════════════════════════════════════════════════════════════════════
# RSA-PSS AUTH
# ══════════════════════════════════════════════════════════════════════════════
# Load private key ONCE at startup (was re-reading from disk on every API call)
_PRIVATE_KEY = None
def _get_private_key():
global _PRIVATE_KEY
if _PRIVATE_KEY is None:
with open(KEY_PATH, "rb") as f:
_PRIVATE_KEY = serialization.load_pem_private_key(f.read(), password=None)
return _PRIVATE_KEY
def _sign(method, path):
ts_ms = str(int(time.time() * 1000))
msg = (ts_ms + method.upper() + path).encode()
pk = _get_private_key()
sig = pk.sign(msg, padding.PSS(mgf=padding.MGF1(hashes.SHA256()),
salt_length=padding.PSS.MAX_LENGTH), hashes.SHA256())
return {"KALSHI-ACCESS-KEY": KEY_ID, "KALSHI-ACCESS-TIMESTAMP": ts_ms,
"KALSHI-ACCESS-SIGNATURE": base64.b64encode(sig).decode(),
"Content-Type": "application/json"}
def api_get(path):
full = "/trade-api/v2" + path
sign_path = full.split("?")[0] # sign without query params
_rate_limit_wait(HOST + full)
r = requests.get(HOST + full, headers=_sign("GET", sign_path), timeout=15)
r.raise_for_status(); return r.json()
def api_post(path, body):
full = "/trade-api/v2" + path
_rate_limit_wait(HOST + full)
r = requests.post(HOST + full, headers=_sign("POST", full), json=body, timeout=15)
if r.status_code >= 400:
try:
detail = r.json()
except Exception:
detail = r.text[:300]
print(f"[api_post] {path} → HTTP {r.status_code}: {detail}")
print(f"[api_post] request body: {body}")
r.raise_for_status(); return r.json()
def api_delete(path):
full = "/trade-api/v2" + path
_rate_limit_wait(HOST + full)
r = requests.delete(HOST + full, headers=_sign("DELETE", full), timeout=15)
if r.status_code not in (200, 204):
print(f"[api_delete] {path} → HTTP {r.status_code}")
return r
def get_portfolio():
resp = api_get("/portfolio/balance")
return resp.get("balance", 0), resp.get("portfolio_value", 0)
# ══════════════════════════════════════════════════════════════════════════════
# ORDER PRUNING
# ══════════════════════════════════════════════════════════════════════════════
def prune_stale_orders():
"""Cancel stale bot-generated orders only. Never cancels manual/external orders."""
try:
orders = api_get("/portfolio/orders?status=resting&limit=100").get("orders", [])
except Exception as e:
print(f"[prune] Could not fetch orders: {e}"); return 0
now = datetime.now(timezone.utc)
cancelled = 0
for o in orders:
oid = o.get("order_id", "")
created_str = o.get("created_time") or o.get("created_at") or ""
client_id = o.get("client_order_id", "")
if not oid or not created_str: continue
# SAFETY: only prune orders placed by this bot (identified by client_order_id prefix)
if not client_id.startswith("mm_"):
continue
try:
created = datetime.fromisoformat(created_str.replace("Z", "+00:00"))
age_h = (now - created).total_seconds() / 3600
if age_h > ORDER_MAX_AGE_HOURS:
r = api_delete(f"/portfolio/orders/{oid}")
if r.status_code in (200, 204):
print(f"[prune] Cancelled {oid} ({o.get('ticker')}, {age_h:.1f}h old)")
cancelled += 1
except Exception as e:
print(f"[prune] Error: {e}")
print(f"[prune] Cancelled {cancelled} stale orders") if cancelled else print("[prune] No stale orders")
return cancelled
# ══════════════════════════════════════════════════════════════════════════════
# POSITION MANAGEMENT — graduated exit logic with health scoring
# ══════════════════════════════════════════════════════════════════════════════
def _init_position_health_table(conn):
"""Create position_health_log table for bandit learning on exit decisions."""
with db_write_ctx(conn):
conn.execute("""CREATE TABLE IF NOT EXISTS position_health_log (
id INTEGER PRIMARY KEY AUTOINCREMENT,
timestamp TEXT NOT NULL,
ticker TEXT NOT NULL,
side TEXT NOT NULL,
quantity INTEGER,
health_score REAL,
remaining_edge REAL,
edge_trend REAL,
action TEXT, -- hold, graduated_trim, graduated_half, etc.
exit_qty INTEGER,
settlement_result TEXT DEFAULT NULL, -- filled post-settlement for learning
settlement_pnl_cents INTEGER DEFAULT NULL
)""")
# Idempotent column adds — exit-model + calibration analysis need fresh_prob
# and source-count trajectories, not just the composite edge snapshot.
for col, coltype in (
("fresh_prob", "REAL"),
("fresh_source_count", "INTEGER"),
("entry_edge", "REAL"),
):
try:
conn.execute(f"ALTER TABLE position_health_log ADD COLUMN {col} {coltype}")
except Exception:
pass # column exists
def _log_position_health(conn, ticker, side, quantity, health,
remaining_edge, trend, action, exit_qty,
fresh_prob=None, fresh_source_count=None,
entry_edge=None):
"""Log a position health evaluation for future bandit learning.
fresh_prob / fresh_source_count capture the ensemble's current FV so the
trajectory can be reconstructed per position. entry_edge is the edge at
the time we opened the position — needed by the edge-decay exit model
(task 4) to compute remaining_edge / entry_edge ratios.
"""
try:
with db_write_ctx(conn):
conn.execute("""INSERT INTO position_health_log
(timestamp, ticker, side, quantity, health_score, remaining_edge,
edge_trend, action, exit_qty,
fresh_prob, fresh_source_count, entry_edge)
VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""",
(datetime.now(timezone.utc).isoformat(), ticker, side, quantity,
round(health, 4), round(remaining_edge, 4),
round(trend, 4), action, exit_qty,
round(fresh_prob, 4) if fresh_prob is not None else None,
fresh_source_count,
round(entry_edge, 4) if entry_edge is not None else None))
except Exception as e:
print(f"[health] log_position_health failed: {e}")
def manage_positions(conn, dynamic_sizing=None):
"""Check existing positions for exit signals. Returns count of exits attempted."""
try:
resp = api_get("/portfolio/positions?limit=100")
positions = resp.get("market_positions", resp.get("positions", []))
except Exception as e:
print(f"[positions] Could not fetch positions: {e}"); return 0
if not positions:
print("[positions] No open positions"); return 0
# Legacy MM positions: MM code has been deleted but mm_inventory rows may still
# hold contracts from the pre-deletion era. Those positions now fall through to
# manage_positions() like any other directional holding — they settle naturally.
# The mm_inventory table is retained for historical analytics only.
now = datetime.now(timezone.utc)
exits = 0
for pos in positions:
ticker = pos.get("ticker", "")
pos_raw = pos.get("position_fp") or pos.get("position", 0)
pos_val = round(float(pos_raw)) if pos_raw else 0
side = "yes" if pos_val > 0 else "no"
quantity = abs(pos_val)
if quantity == 0: continue
# Get current market price for this position
try:
mkt = api_get(f"/markets/{ticker}")
market = mkt.get("market", mkt)
except Exception as e:
print(f"[positions] Could not fetch market {ticker}: {e}"); continue
# Current price
yes_ask = float(market.get("yes_ask") or market.get("yes_ask_dollars") or 0)
yes_bid = float(market.get("yes_bid") or market.get("yes_bid_dollars") or 0)
no_ask = float(market.get("no_ask") or market.get("no_ask_dollars") or 0)
no_bid = float(market.get("no_bid") or market.get("no_bid_dollars") or 0)
if yes_ask > 1: yes_ask /= 100
if yes_bid > 1: yes_bid /= 100
if no_ask > 1: no_ask /= 100
if no_bid > 1: no_bid /= 100
# Current exit price (what we'd get selling now) = bid on our side
exit_price = yes_bid if side == "yes" else no_bid
if exit_price <= 0: continue
# Look up entry price + entry edge from our trades DB. entry_edge
# feeds the edge-decay exit model (task 4) and lets the bandit
# learn remaining_edge / entry_edge ratios at settlement.
trade = conn.execute(
"SELECT price_cents, timestamp, edge FROM trades "
"WHERE ticker=? AND side=? AND action='buy' ORDER BY id DESC LIMIT 1",
(ticker, side)
).fetchone()
entry_edge = None
if trade:
entry_price = trade[0] / 100 # cents to fraction
entry_time = trade[1]
entry_edge = trade[2] # float or None
else:
# Fallback: use position's average cost if available
entry_price = float(pos.get("average_price_paid") or pos.get("market_exposure") or 0)
if entry_price > 1: entry_price /= 100
entry_time = None
if entry_price <= 0: continue
# Calculate P&L net of estimated round-trip fees
# Entry fee + exit fee + exit spread ≈ 2-3% for typical positions
estimated_round_trip_fee = 0.02 * entry_price + 0.01 # ~2% of entry + 1¢ spread
pnl_pct = (exit_price - entry_price - estimated_round_trip_fee) / entry_price
# Check holding period
days_held = 0
if entry_time:
try:
entry_dt = datetime.fromisoformat(entry_time.replace("Z", "+00:00"))
days_held = (now - entry_dt).total_seconds() / 86400
except Exception:
pass
# ── Graduated exit decision ───────────────────────────────────────
# Instead of binary hold/exit, compute a POSITION HEALTH SCORE (0-1)
# and map it to graduated actions: hold / trim / partial exit / full exit.
#
# Health = f(remaining_edge, edge_trend, time_pressure, source_confidence)
# This replaces the static 3¢ flip threshold with a dynamic, fee-aware model.
# Dynamic thresholds scale with total equity (min contracts to partial exit)
_trim_thresh = (dynamic_sizing or {}).get("trim_threshold", 3)
_major_thresh = (dynamic_sizing or {}).get("major_threshold", 5)
exit_reason = None
exit_qty = quantity # default: full exit
exit_urgency = 0 # 0=patient, 1=moderate, 2=aggressive pricing
health = 0.0
remaining_edge = 0.0
trend_score = 0.0
# 1. Hard triggers (always full exit, no health calc needed)
if pnl_pct >= TAKE_PROFIT_PCT:
exit_reason = f"take_profit: +{pnl_pct:.0%} (>{TAKE_PROFIT_PCT:.0%})"
exit_urgency = 0 # patient — we're winning
elif pnl_pct <= -STOP_LOSS_PCT:
exit_reason = f"stop_loss: {pnl_pct:.0%} (<-{STOP_LOSS_PCT:.0%})"
exit_urgency = 2 # aggressive — bleeding
# 2. Graduated health-based exit (replaces static ensemble_flip)
if not exit_reason:
fresh_prob = None
fresh_n = 0
fresh_src = ""
try:
current_ask = yes_ask if side == "yes" else no_ask
vol = float(market.get("volume") or market.get("volume_fp") or 0)
fresh_prob, fresh_src, fresh_n = get_independent_estimate(
ticker, market, current_ask, vol)
except Exception as e:
print(f"[positions] {ticker}: ensemble re-eval failed: {e}")
# Compute remaining edge (net of exit fees)
# For YES position: edge = fresh_prob - exit_price (positive = still profitable)
# For NO position: edge = (1-fresh_prob) - (1-yes_bid)
exit_fee_est = 0.015 # ~1.5¢ maker exit fee estimate
if fresh_prob is not None and fresh_n > 0:
if side == "yes":
remaining_edge = fresh_prob - exit_price - exit_fee_est
else:
remaining_edge = (1 - fresh_prob) - (1 - yes_bid) - exit_fee_est
else:
# No fresh data — use P&L as proxy (if losing, assume edge gone)
remaining_edge = pnl_pct * entry_price
# Time pressure: less time = need stronger edge to justify holding
# Expiry check
close_str = (market.get("close_time") or market.get("expiration_time") or "")
hours_to_expiry = 999
if close_str:
try:
close_dt = datetime.fromisoformat(close_str.replace("Z", "+00:00"))
hours_to_expiry = max(0, (close_dt - now).total_seconds() / 3600)
except Exception:
pass
# ── Settlement certainty fast-path ──────────────────────────────
# When the ensemble strongly indicates the outcome is locked in AND
# we're near expiry, bypass the health score entirely.
#
# Two cases:
# 1. Near-certain LOSER: exit aggressively to salvage remaining value
# 2. Near-certain WINNER: hold to settlement (no point selling at 90¢
# and paying fees when settlement pays $1)
_SETTLEMENT_CERTAINTY_THRESH = 0.90 # 90% confidence
_SETTLEMENT_CERTAINTY_HOURS = 4 # within 4 hours of settlement
_SETTLEMENT_HOLD_THRESH = 0.93 # 93% confidence to hold to settlement
_SETTLEMENT_HOLD_HOURS = 2 # within 2 hours of settlement
if (fresh_prob is not None and fresh_n > 0
and hours_to_expiry < _SETTLEMENT_CERTAINTY_HOURS):
our_prob = fresh_prob if side == "yes" else (1 - fresh_prob)
if our_prob <= (1 - _SETTLEMENT_CERTAINTY_THRESH):
# Our side is almost certainly LOSING — exit immediately
exit_reason = (
f"settlement_certainty_exit: P(our_side)={our_prob:.2f} "
f"({fresh_n} sources) hrs_left={hours_to_expiry:.1f}h "
f"— near-certain loser, salvaging remaining value"
)
exit_urgency = 2 # aggressive — cross the spread
health = our_prob # record for logging
print(f"[positions] 🚨 {ticker} {side} x{quantity}: "
f"SETTLEMENT CERTAINTY EXIT — our_prob={our_prob:.2f} "
f"hrs_left={hours_to_expiry:.1f}h")
elif (our_prob >= _SETTLEMENT_HOLD_THRESH
and hours_to_expiry < _SETTLEMENT_HOLD_HOURS):
# Our side is almost certainly WINNING — hold to settlement
# Don't sell at 90¢+ and pay fees when $1 is coming at settlement.
print(f"[positions] ✅ {ticker} {side} x{quantity}: "
f"entry={entry_price:.2f} now={exit_price:.2f} pnl={pnl_pct:+.0%} "
f"P(our_side)={our_prob:.2f} hrs_left={hours_to_expiry:.1f}h "
f"— SETTLEMENT CERTAINTY HOLD")
_log_position_health(conn, ticker, side, quantity,
our_prob, remaining_edge, 0.0,
"settlement_hold", 0,
fresh_prob=fresh_prob,
fresh_source_count=fresh_n,
entry_edge=entry_edge)
continue # skip to next position
# Time decay multiplier: require more edge as expiry approaches
if hours_to_expiry < 2:
time_mult = 2.0 # need 2x normal edge
elif hours_to_expiry < 12:
time_mult = 1.5
elif hours_to_expiry < 48:
time_mult = 1.2
else:
time_mult = 1.0
# Edge trend: check if edge has been deteriorating
# Use kv_cache to track edge history across cycles
edge_trend_key = f"pos_edge_{ticker}_{side}"
edge_history = _db_cache_get(conn, edge_trend_key) or []
edge_history.append({"edge": round(remaining_edge, 4), "ts": time.time()})
# Keep last 10 readings (20 minutes at 2-min cycles)
edge_history = edge_history[-10:]
_db_cache_set(conn, edge_trend_key, edge_history, 7200) # 2h TTL
# Trend: compare avg of last 3 readings to avg of first 3
trend_score = 0.0 # -1=deteriorating, 0=flat, +1=improving
if len(edge_history) >= 6:
recent = [h["edge"] for h in edge_history[-3:]]
earlier = [h["edge"] for h in edge_history[:3]]
avg_recent = sum(recent) / len(recent)
avg_earlier = sum(earlier) / len(earlier)
if avg_earlier != 0:
trend_score = max(-1, min(1, (avg_recent - avg_earlier) / max(abs(avg_earlier), 0.01)))
# ── Edge-decay hard trigger ─────────────────────────────────
# Runs before the composite health score when entry_edge is
# available. Inverted-thesis and decayed-below-floor exits are
# the semantic anchor; time + stale backstops cover frozen-
# ensemble failure modes. Composite health-score still applies
# when entry_edge is missing (older positions, shadow-only).
if entry_edge is not None and fresh_prob is not None and fresh_n > 0:
from bot.core.exit_model import evaluate_edge_decay_exit
from bot.config import (
EXIT_EDGE_DECAY_RATIO, EXIT_TIME_BACKSTOP_HOURS,
EXIT_TIME_BACKSTOP_EDGE_ABS, EXIT_STALE_HOLD_HOURS,
)
_decay_dec = evaluate_edge_decay_exit(
entry_edge=entry_edge,
remaining_edge=remaining_edge,
hours_to_expiry=hours_to_expiry,
hours_held=days_held * 24.0,
trend_score=trend_score,
decay_ratio=EXIT_EDGE_DECAY_RATIO,
time_backstop_hours=EXIT_TIME_BACKSTOP_HOURS,
time_backstop_edge_abs=EXIT_TIME_BACKSTOP_EDGE_ABS,
stale_hold_hours=EXIT_STALE_HOLD_HOURS,
)
if _decay_dec.trigger is not None:
exit_reason = f"edge_decay_{_decay_dec.trigger}: {_decay_dec.detail}"
# Urgency: flipped edge is most urgent (thesis inverted),
# decayed is moderate, backstops patient.
exit_urgency = (2 if _decay_dec.trigger == "edge_flipped"
else 1 if _decay_dec.trigger == "edge_decayed"
else 0)
# Source confidence factor: more sources = more trust in the estimate
confidence = min(1.0, fresh_n / 3.0) if fresh_n > 0 else 0.3
# ── POSITION HEALTH SCORE (0-1) ──
# Components:
# edge_component: is remaining edge positive and sufficient?
# trend_component: is edge improving or deteriorating?
# time_component: penalty for holding with slim edge near expiry
# pnl_component: unrealized P&L as a sanity check
min_edge_required = 0.02 * time_mult # ~2¢ base, scaled by time pressure
edge_component = max(0, min(1, (remaining_edge + 0.05) / 0.10)) # 0 at -5¢, 1 at +5¢
trend_component = (trend_score + 1) / 2 # normalize to 0-1
time_component = min(1.0, hours_to_expiry / 48) # 0 at expiry, 1 at 48h+
pnl_component = max(0, min(1, (pnl_pct + STOP_LOSS_PCT) / (TAKE_PROFIT_PCT + STOP_LOSS_PCT)))
# Weighted health score
health = (0.40 * edge_component +
0.20 * trend_component +
0.15 * time_component +
0.15 * pnl_component +
0.10 * confidence)
# ── Map health to action ──
# Edge-decay trigger (above) may have already set exit_reason.
# When it has, skip the health-score branching so we exit at
# the semantically-anchored reason rather than re-deriving from
# the composite — and exit_qty stays at the default (full exit).
if exit_reason:
pass # fall through to exit-execution below
elif health >= 0.65:
# Healthy — hold
print(f"[positions] {ticker} {side} x{quantity}: "
f"entry={entry_price:.2f} now={exit_price:.2f} pnl={pnl_pct:+.0%} "
f"edge={remaining_edge:+.3f} health={health:.2f} "
f"trend={'↑' if trend_score > 0.2 else '↓' if trend_score < -0.2 else '→'} "
f"held={days_held:.1f}d — HOLD")
# Log health for learning
_log_position_health(conn, ticker, side, quantity, health,
remaining_edge, trend_score, "hold", 0,
fresh_prob=fresh_prob,
fresh_source_count=fresh_n,
entry_edge=entry_edge)
continue
elif health >= 0.45 and quantity > _trim_thresh:
# Moderate — trim 25-33% (only if position is large enough)
trim_pct = 0.25 if health >= 0.55 else 0.33
exit_qty = max(1, int(quantity * trim_pct))
exit_reason = (f"graduated_trim: health={health:.2f} edge={remaining_edge:+.3f} "
f"trend={trend_score:+.2f} → selling {exit_qty}/{quantity}")
exit_urgency = 0
elif health >= 0.30:
# Weak — exit 50% if large enough, full exit if small
if quantity > _trim_thresh:
exit_qty = max(1, quantity // 2)
exit_reason = (f"graduated_half: health={health:.2f} edge={remaining_edge:+.3f} "
f"trend={trend_score:+.2f} → selling {exit_qty}/{quantity}")
else:
exit_reason = (f"graduated_exit: health={health:.2f} edge={remaining_edge:+.3f}")
exit_urgency = 1
elif health >= 0.15:
# Poor — exit 75% of large positions, full exit of small
if quantity > _major_thresh:
exit_qty = max(1, int(quantity * 0.75))
exit_reason = (f"graduated_major: health={health:.2f} edge={remaining_edge:+.3f} "
f"trend={trend_score:+.2f} → selling {exit_qty}/{quantity}")
else:
exit_reason = (f"graduated_exit: health={health:.2f} edge={remaining_edge:+.3f}")
exit_urgency = 1
else:
# Critical — full exit
exit_reason = (f"graduated_critical: health={health:.2f} edge={remaining_edge:+.3f} "
f"trend={trend_score:+.2f}")
exit_urgency = 2
# Time-based exit still fires as a backstop
if not exit_reason and days_held >= MAX_HOLD_DAYS:
exit_reason = f"time_exit: held {days_held:.1f} days (>{MAX_HOLD_DAYS}d)"
exit_urgency = 1
if not exit_reason:
print(f"[positions] {ticker} {side} x{quantity}: "
f"entry={entry_price:.2f} now={exit_price:.2f} pnl={pnl_pct:+.0%} "
f"held={days_held:.1f}d — HOLD")
continue
# Exit escalation: if we've been trying patient exits for 2+ cycles
# without the position clearing, escalate urgency.
exit_attempt_key = f"exit_attempt_{ticker}_{side}"
prior_attempts = _db_cache_get(conn, exit_attempt_key)
if prior_attempts and isinstance(prior_attempts, dict):
attempt_count = prior_attempts.get("count", 0) + 1
if exit_urgency == 0 and attempt_count >= 2:
exit_urgency = 1
print(f"[positions] {ticker}: escalating exit urgency to 1 "
f"(patient exit pending {attempt_count} cycles)")
elif exit_urgency == 1 and attempt_count >= 4:
exit_urgency = 2
print(f"[positions] {ticker}: escalating exit urgency to 2 "
f"(moderate exit pending {attempt_count} cycles)")
_db_cache_set(conn, exit_attempt_key, {"count": attempt_count}, 7200)
else:
_db_cache_set(conn, exit_attempt_key, {"count": 1}, 7200)
print(f"[positions] {ticker} {side} x{exit_qty}/{quantity}: "
f"entry={entry_price:.2f} now={exit_price:.2f} pnl={pnl_pct:+.0%} — EXIT: {exit_reason}")
# Synthetic sell: BUY the opposite side instead of selling.
# Maker fee (~0.44¢/contract) vs taker fee (~1.75¢/contract) = 4x savings.
# For YES position: buy NO at aggressive price. For NO position: buy YES.
opp_side = "no" if side == "yes" else "yes"
if side == "yes":
# Exiting YES → buy NO. NO price = 100 - yes_bid.
# Urgency adjusts how much above ask we're willing to pay.
# round(), not int(): avoids 1¢ misprice from float underflow
# (CLAUDE.md §5 — fixed-point parsing).
base_no_price = max(1, 100 - round(yes_bid * 100))
if exit_urgency >= 2:
opp_price_cents = min(99, base_no_price + 3)
elif exit_urgency == 1:
opp_price_cents = min(99, base_no_price + 1)
else:
opp_price_cents = min(99, base_no_price)
else:
# Exiting NO → buy YES. YES price = 100 - no_bid.
base_yes_price = max(1, 100 - round(no_bid * 100))
if exit_urgency >= 2:
opp_price_cents = min(99, base_yes_price + 3)
elif exit_urgency == 1:
opp_price_cents = min(99, base_yes_price + 1)
else:
opp_price_cents = min(99, base_yes_price)
order_id = None
error = None
if not DRY_RUN:
try:
# mm_exit_ prefix: fills writer's source_tagger routes this to
# `exit`. Periods stripped per Kalshi constraint (CLAUDE.md
# §Known Bug Pattern #1).
client_id = f"mm_exit_{ticker.replace('.', '_')}_{int(time.time())}"
order_body = {
"ticker": ticker,
"side": opp_side,
"type": "limit",
"count": exit_qty,
("yes_price" if opp_side == "yes" else "no_price"): opp_price_cents,
"action": "buy",
"expiration_ts": int(time.time() + ORDER_MAX_AGE_HOURS * 3600),
"client_order_id": client_id,
}
# Urgency 0: patient exit — post_only to get maker fee, rests at ask
# Urgency 1+: must cross the spread to fill — no post_only
# (post_only + crossing price = Kalshi rejects the order silently)
if exit_urgency == 0:
order_body["post_only"] = True
resp = api_post("/portfolio/orders", order_body)
order_id = resp.get("order", {}).get("order_id") or str(resp)
mode = "maker" if exit_urgency == 0 else "taker"
print(f" ✓ synthetic sell: buy {opp_side} {exit_qty}x @ {opp_price_cents}¢ "
f"(urgency={exit_urgency}, {mode})")
except Exception as e:
error = str(e)
print(f" ✗ synthetic sell failed: {error}")
else:
print(f" [DRY RUN] would buy {opp_side} {exit_qty}x {ticker} @ {opp_price_cents}¢ (synthetic sell)")
# Log for bandit learning
_log_position_health(conn, ticker, side, exit_qty,
health, remaining_edge, trend_score,
exit_reason.split(":")[0], exit_qty,
fresh_prob=fresh_prob,
fresh_source_count=fresh_n,
entry_edge=entry_edge)
with db_write_ctx(conn):
conn.execute("""INSERT INTO position_exits
(timestamp, ticker, side, entry_price_cents, exit_price_cents,
contracts, exit_reason, order_id, error)
VALUES (?,?,?,?,?,?,?,?,?)""",
(now.isoformat(), ticker, side, round(entry_price * 100),
opp_price_cents, quantity, exit_reason, order_id, error))
exits += 1
print(f"[positions] {exits} exit orders placed")
return exits
# ══════════════════════════════════════════════════════════════════════════════
# INFORMATION LAYER — independent probability estimates from external sources
# ══════════════════════════════════════════════════════════════════════════════
import re, math
from cachetools import TTLCache
# Shared cache: {key: (value, timestamp)} — bounded TTLCache so the long-running
# daemon can't OOM via unbounded growth (CLAUDE.md Known Bug Pattern #12).
# maxsize is generous; ttl here matches the longest manual-TTL check below
# (60-min vol cache) so the cachetools layer is a backstop, not the primary
# eviction mechanism (call sites still do their own time-since-write check).
CACHE_TTL = 60 # seconds — primary TTL used by manual checks at call sites
_CACHE = TTLCache(maxsize=4096, ttl=3600)
# ── Persistent cache (SQLite) — canonical implementation in bot/db ──────
import json as _json_mod # kept for other uses in trade.py
from bot.db import kv_get as _db_cache_get, kv_set as _db_cache_set, kv_cleanup as _db_cache_cleanup, db_write_ctx # noqa: E402
# ── Rate limiter: per-domain minimum interval between requests ───────────
_RATE_LIMITS = {
# domain substring → (min_interval_seconds, max_burst)
"kalshi": (0.25, 8), # 4 req/s max, burst of 8
"polymarket": (0.5, 4), # 2 req/s, burst of 4
"open-meteo": (1.0, 3), # 1 req/s (free tier)
"coingecko": (1.5, 2), # free tier strict
"coingecko": (1.5, 2), # alternate spelling
"fred.stlouisfed": (1.0, 3), # FRED free tier
"the-odds-api": (2.0, 2), # precious credits — go slow
"metaculus": (1.0, 3),
"finnhub": (1.0, 3),
"deribit": (1.0, 3),
"noaa": (1.0, 3),
"clevelandfed": (2.0, 2),
"openai": (0.5, 5), # GPT-4o-mini calls
"manifold": (0.5, 4), # Manifold Markets API (generous limits)
"bls.gov": (2.0, 3), # BLS API v2 (500 req/day with key)
"tomorrow.io": (2.0, 3), # Tomorrow.io (500 calls/day)
}
# Tracks {domain_key: [timestamp_of_recent_requests]}
_RATE_HISTORY = {}
def _rate_limit_wait(url):
"""Enforce per-domain rate limiting. Blocks until it's safe to make the request."""
from urllib.parse import urlparse
domain = urlparse(url).hostname or ""
matched_key = None
for key in _RATE_LIMITS:
if key in domain:
matched_key = key
break
if not matched_key:
return # No rate limit configured for this domain
min_interval, max_burst = _RATE_LIMITS[matched_key]
now = time.time()
if matched_key not in _RATE_HISTORY:
_RATE_HISTORY[matched_key] = []
history = _RATE_HISTORY[matched_key]
# Prune old entries (older than max_burst * min_interval)
window = max_burst * min_interval
history[:] = [t for t in history if now - t < window]
if len(history) >= max_burst:
# We've hit burst limit — wait until oldest request exits the window
wait_until = history[0] + window
sleep_time = wait_until - now
if sleep_time > 0:
time.sleep(sleep_time)
elif history:
# Enforce minimum interval since last request
time_since_last = now - history[-1]
if time_since_last < min_interval:
time.sleep(min_interval - time_since_last)
_RATE_HISTORY[matched_key].append(time.time())
_DEFAULT_HEADERS = {
"User-Agent": "KalshiTradingBot/1.0 (contact: bot@example.com)",
"Accept": "application/json",
}
def _cached_get(key, url, timeout=5, headers=None):
"""GET with in-memory cache, per-domain rate limiting, and retry on transient errors."""
now = time.time()
if key in _CACHE and now - _CACHE[key][1] < CACHE_TTL:
return _CACHE[key][0]
if not url:
return None # guard against None URLs
max_retries = 2
for attempt in range(max_retries + 1):
try:
_rate_limit_wait(url)
hdrs = {**_DEFAULT_HEADERS, **(headers or {})}
r = requests.get(url, timeout=timeout, headers=hdrs)
if r.status_code in (500, 502, 503) and attempt < max_retries:
time.sleep(1.0 * (attempt + 1)) # backoff: 1s, 2s
continue
if r.status_code != 200:
print(f"[http] {key} → HTTP {r.status_code} from {url.split('?')[0]}")
return None