"""BTC 15-minute "Up or Down" strategy, running continuously. Every 15-minute window: 1. wait for the open KXBTC15M window 2. sleep until MINUTES_LEFT before close 3. if the YES or NO price is between MIN_PRICE and MAX_PRICE, buy CONTRACTS of that side with a limit order at that price 4. cancel whatever has not filled shortly before close 5. wait for settlement and log the result to btc_15m_trades.csv With DRY_RUN = True nothing is bought: the order is priced with a preflight and the result is logged as if it had filled. Set API_SECRET_KEY and DEFAULT_ACCOUNT_NUMBER as environment variables or in a .env file next to this script, then run it; stop with Ctrl-C: python3 btc_15m_strategy.py """ import csv import logging import os import time from datetime import datetime, timedelta, timezone from decimal import Decimal from pathlib import Path from typing import Optional, Tuple from dotenv import load_dotenv from public_api_sdk import ( ApiKeyAuthConfig, EventContract, EventOutcome, OrderStatus, PublicApiClient, PublicApiClientConfiguration, WaitTimeoutError, ) load_dotenv() # === Settings ================================================================ DRY_RUN = True # True: only log what it would buy. False: place real orders. MINUTES_LEFT = 5 # check prices when this many minutes are left in the window MIN_PRICE = 0.80 # buy YES or NO when its price is at least this (0.80 = 80%) MAX_PRICE = 0.95 # ...but no higher than this (fees are ~$0.02 per contract) CONTRACTS = 1 # contracts to buy per window # === Internals (no need to change) =========================================== SERIES = "KXBTC15M" CATEGORY, SUBCATEGORY = "Crypto", "BTC" ENTRY_GRACE_SECONDS = 1100 # skip a window if we reach it later than this CANCEL_SECONDS_BEFORE_CLOSE = 5 # cancel an unfilled order this close to close SETTLE_TIMEOUT_SECONDS = 600 # windows settle ~5 min after close THRESHOLD = Decimal(str(MIN_PRICE)) PRICE_CAP = Decimal(str(MAX_PRICE)) QUANTITY = Decimal(str(CONTRACTS)) HERE = Path(__file__).resolve().parent TRADES_CSV = HERE / "btc_15m_trades.csv" LOG_FILE = HERE / "btc_15m_strategy.log" logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.StreamHandler(), logging.FileHandler(LOG_FILE)], ) log = logging.getLogger("btc_15m") # --- Helpers ---------------------------------------------------------------- def now() -> datetime: return datetime.now(timezone.utc) def sleep_until(when: datetime) -> None: while (remaining := (when - now()).total_seconds()) > 0: time.sleep(min(remaining, 30)) def pick_side(outcome: EventOutcome) -> Optional[EventContract]: """The YES or NO contract whose ask is between MIN_PRICE and MAX_PRICE. The SDK buys at the ask rounded up to a whole cent, which stays within MAX_PRICE as long as MAX_PRICE is in whole cents. """ sides = [ c for c in (outcome.yes, outcome.no) if c is not None and c.ask is not None and THRESHOLD <= c.ask <= PRICE_CAP ] return max(sides, key=lambda c: c.ask, default=None) def already_holding(client: PublicApiClient, outcome: EventOutcome) -> bool: """True if we hold either contract of this outcome (e.g. after a restart).""" symbols = {c.symbol for c in outcome.contracts} return any(p.instrument.symbol in symbols for p in client.get_portfolio().positions) def record(row: dict) -> None: new_file = not TRADES_CSV.exists() with TRADES_CSV.open("a", newline="") as f: writer = csv.DictWriter(f, fieldnames=list(row)) if new_file: writer.writeheader() writer.writerow(row) # --- One window -------------------------------------------------------------- def buy( client: PublicApiClient, contract: EventContract, outcome: EventOutcome ) -> Tuple[Decimal, Decimal, Decimal]: """Buy QUANTITY of `contract` at its ask. Returns (filled quantity, cost incl. fees, limit price). """ preview = client.preflight_buy_event_contract(contract, QUANTITY) cost = preview.estimated_cost or preview.order_value price = (preview.order_value / QUANTITY).quantize(Decimal("0.01")) if DRY_RUN: log.info("DRY RUN: would buy %s x %s @ %s (est. cost %s)", QUANTITY, contract.symbol, price, cost) return QUANTITY, cost, price order = client.buy_event_contract(contract, QUANTITY) log.info("Placed order %s: buy %s x %s @ %s", order.order_id, QUANTITY, contract.symbol, price) timeout = (outcome.seconds_left or 0) - CANCEL_SECONDS_BEFORE_CLOSE try: filled = order.wait_for_fill(timeout=max(timeout, 1)) except WaitTimeoutError: log.info("Not filled before close; cancelling %s", order.order_id) order.cancel() try: filled = order.wait_for_terminal_status(timeout=15) except WaitTimeoutError: filled = order.get_details() qty = filled.filled_quantity or Decimal(0) if filled.status != OrderStatus.FILLED: log.info("Order %s ended %s with %s filled", order.order_id, filled.status, qty) # scale the preflight cost to the quantity that actually filled return qty, cost * qty / QUANTITY, price def run_window(client: PublicApiClient) -> None: log.info("Waiting for the next %s window", SERIES) event = client.wait_for_current_event( SERIES, category=CATEGORY, subcategory=SUBCATEGORY ) outcome = client.get_event_details(event.event_symbol).outcome if outcome is None: log.warning("%s does not have exactly one outcome; skipping", event.event_symbol) sleep_until(event.resolution_time + timedelta(seconds=5)) return close_time = outcome.close_time or event.resolution_time entry_time = close_time - timedelta(minutes=MINUTES_LEFT) log.info("%s (%s) closes %s; entry check at %s", event.event_symbol, outcome.title, close_time, entry_time) if now() > entry_time + timedelta(seconds=ENTRY_GRACE_SECONDS): log.info("Too late for this window; waiting for the next one") sleep_until(close_time + timedelta(seconds=5)) return sleep_until(entry_time) outcome = client.get_event_details(event.event_symbol).outcome assert outcome is not None yes, no = outcome.yes, outcome.no log.info("Entry check: tradable=%s, YES bid/ask %s/%s, NO bid/ask %s/%s", outcome.is_tradable, yes and yes.bid, yes and yes.ask, no and no.bid, no and no.ask) row = { "time": now().isoformat(timespec="seconds"), "event": event.event_symbol, "target": outcome.title, "yes_bid_ask": (yes.bid, yes.ask) if yes else None, "no_bid_ask": (no.bid, no.ask) if no else None, "action": "skip", "contract": "", "price": "", "quantity": "", "cost": "", "result": "", "pnl": "", "dry_run": DRY_RUN, } contract = pick_side(outcome) if not outcome.is_tradable: row["action"] = "skip: not tradable" elif contract is None: row["action"] = f"skip: no price in [{THRESHOLD}, {PRICE_CAP}]" elif not DRY_RUN and already_holding(client, outcome): row["action"] = "skip: already holding" else: qty, cost, price = buy(client, contract, outcome) row.update(action="buy" if qty else "buy: unfilled", contract=contract.symbol, price=price, quantity=qty, cost=cost) if qty: try: result = client.wait_for_settlement( event.event_symbol, timeout=SETTLE_TIMEOUT_SECONDS ) except WaitTimeoutError: row["result"] = "not settled yet" else: row["result"] = result.settled_outcome.value if result.settled_outcome else "" won = result.won(contract) if won is not None: row["pnl"] = (qty if won else Decimal(0)) - cost log.info("Settled %s → %s, P&L %s", event.event_symbol, row["result"], row["pnl"]) log.info("Window done: %s", row["action"]) record(row) sleep_until(close_time + timedelta(seconds=5)) def main() -> None: client = PublicApiClient( ApiKeyAuthConfig(api_secret_key=os.environ["API_SECRET_KEY"]), config=PublicApiClientConfiguration( default_account_number=os.environ["DEFAULT_ACCOUNT_NUMBER"] ), ) log.info("Starting %s strategy (DRY_RUN=%s, min %s, max %s, qty %s)", SERIES, DRY_RUN, THRESHOLD, PRICE_CAP, QUANTITY) try: while True: try: run_window(client) except KeyboardInterrupt: raise except Exception: log.exception("Window failed; retrying in 15s") time.sleep(15) except KeyboardInterrupt: log.info("Stopped") finally: client.close() if __name__ == "__main__": main()