`assets/bot-template/bot.py`

169 lines
#!/usr/bin/env python3
"""
bot.py — the runner. One cycle:

  1. fetch the market universe (venue_adapter, read-only)
  2. mark + evaluate EXIT rules on every open position → close matches
  3. filter the universe, evaluate ENTRY rules on candidates
  4. apply risk caps (max positions, exposure, daily loss, kill switch)
  5. route surviving decisions to the executor (paper by default)
  6. print a per-cycle summary + an explicit cost line

Usage:
    python bot.py --once              # one cycle, then exit
    python bot.py --loop              # loop forever every strategy.poll_seconds
    python bot.py --once --dry-run    # decide but place NO fills (even paper)
    python bot.py --strategy other.json

Paper is the default and requires no keys. Live mode is gated (see executor.py).
"""

import argparse
import json
import os
import time

import ledger
from executor import get_executor
from safe_eval import MARKET_VARS, eval_rule
from venue_adapter import get_adapter

HERE = os.path.dirname(os.path.abspath(__file__))


def load_strategy(path):
    with open(path) as f:
        s = json.load(f)
    s.setdefault("mode", "paper")
    s.setdefault("poll_seconds", 300)
    s.setdefault("universe", {})
    s.setdefault("sizing", {})
    s.setdefault("risk", {})
    return s


def market_vars(m):
    return {k: m.get(k, 0) for k in MARKET_VARS}


def passes_universe(m, u):
    if m["volume_usd"] < u.get("min_volume_usd", 0):
        return False
    if m["spread"] > u.get("max_spread", 1.0):
        return False
    if m["days_to_resolve"] > u.get("resolves_within_days", 1e9):
        return False
    q = u.get("query", "")
    return not q or q.lower() in m["title"].lower()


def run_cycle(strategy, adapter, executor, dry_run=False):
    mode = strategy["mode"]
    sizing, risk = strategy["sizing"], strategy["risk"]
    actions = []

    # kill switch: a data/KILL file halts all new entries immediately.
    kill = os.path.exists(os.path.join(HERE, "data", "KILL"))

    universe = adapter.fetch_universe(
        query=strategy["universe"].get("query", ""),
        limit=strategy["universe"].get("limit", 200),
    )
    by_id = {m["id"]: m for m in universe}

    # --- exits first (always allowed, even when the kill switch is on) ---
    for pos in ledger.open_positions().values():
        mark = by_id.get(pos["market_id"], {}).get("price")
        if mark is None:
            mark = adapter.mark_price(pos["market_id"])
        if mark is None:
            continue
        mvars = {"price": mark, "volume_usd": 0, "liquidity_usd": 0, "spread": 0,
                 "days_to_resolve": by_id.get(pos["market_id"], {}).get("days_to_resolve", 9999)}
        for rule in strategy.get("exit", []):
            if eval_rule(rule["when"], mvars):
                actions.append(("close", pos["title"], rule["reason"], mark))
                if not dry_run:
                    executor.close_position(pos, mark, rule["reason"])
                break

    # --- daily-loss circuit breaker ---
    max_daily_loss = risk.get("max_daily_loss_usd")
    if max_daily_loss is not None and ledger.pnl_today_usd() <= -abs(max_daily_loss):
        return {"actions": actions, "halted": "max_daily_loss hit — no new entries",
                "summary": ledger.summary(mark_fn=adapter.mark_price)}

    if kill or risk.get("kill_switch_engaged"):
        return {"actions": actions, "halted": "kill switch engaged",
                "summary": ledger.summary(mark_fn=adapter.mark_price)}

    # --- entries ---
    open_now = ledger.open_positions()
    for m in universe:
        if m["id"] in open_now:
            continue
        if len(open_now) >= sizing.get("max_open_positions", 9999):
            break
        if not passes_universe(m, strategy["universe"]):
            continue
        size = min(sizing.get("amount", 0), sizing.get("max_position_usd", 1e9))
        if ledger.total_exposure_usd() + size > risk.get("max_total_exposure_usd", 1e12):
            continue
        for rule in strategy.get("entry", []):
            if eval_rule(rule["when"], market_vars(m)):
                actions.append(("open", m["title"], rule["reason"], m["price"]))
                if not dry_run:
                    executor.open_position(m, rule.get("side", "YES"), size, rule["reason"])
                    open_now[m["id"]] = True
                break

    return {"actions": actions, "halted": None,
            "summary": ledger.summary(mark_fn=adapter.mark_price)}


def print_cycle(strategy, result, dry_run):
    tag = "DRY-RUN" if dry_run else strategy["mode"].upper()
    print(f"\n── cycle [{tag}] {strategy['name']} @ {strategy['venue']} ──")
    if result["halted"]:
        print(f"  ⛔ HALTED: {result['halted']}")
    if not result["actions"]:
        print("  · no entries/exits triggered this cycle")
    for act, title, reason, price in result["actions"]:
        arrow = "▲ OPEN " if act == "open" else "▼ CLOSE"
        print(f"  {arrow} {price:.3f}  {title[:60]}  ({reason})")
    s = result["summary"]
    print(f"  positions={s['open_count']} exposure=${s['exposure_usd']:.0f} "
          f"realized=${s['realized_pnl_usd']:.2f} unrealized=${s['unrealized_pnl_usd']:.2f} "
          f"today=${s['pnl_today_usd']:.2f}")
    cost = "$0 (paper — public API only)" if strategy["mode"] != "live" else "REAL MONEY (live)"
    print(f"  cost: {cost}")


def main():
    ap = argparse.ArgumentParser(description="zero-code prediction-market paper bot")
    ap.add_argument("--strategy", default=os.path.join(HERE, "strategy.json"))
    ap.add_argument("--once", action="store_true", help="run a single cycle")
    ap.add_argument("--loop", action="store_true", help="loop every poll_seconds")
    ap.add_argument("--dry-run", action="store_true", help="decide but place no fills")
    args = ap.parse_args()

    strategy = load_strategy(args.strategy)
    adapter = get_adapter(strategy["venue"])
    executor = get_executor(strategy["mode"], strategy["venue"])

    if args.loop:
        print(f"looping every {strategy['poll_seconds']}s — Ctrl-C to stop "
              f"(touch data/KILL to halt entries)")
        while True:
            try:
                print_cycle(strategy, run_cycle(strategy, adapter, executor, args.dry_run), args.dry_run)
            except Exception as e:  # a bad cycle should never kill the loop
                print(f"  ! cycle error: {e}")
            time.sleep(strategy["poll_seconds"])
    else:
        print_cycle(strategy, run_cycle(strategy, adapter, executor, args.dry_run), args.dry_run)


if __name__ == "__main__":
    main()