|
| 1 | +"""Shared helpers for research-only simulation scripts.""" |
| 2 | +from __future__ import annotations |
| 3 | + |
| 4 | +import statistics |
| 5 | +from typing import Any |
| 6 | + |
| 7 | +DEFAULT_TICK_SIZE = 0.001 |
| 8 | +EPSILON = 1e-9 |
| 9 | + |
| 10 | + |
| 11 | +def simulate_buy_cost(asks: list[tuple[float, float]], target_units: float) -> tuple[float, float, float]: |
| 12 | + """Walk an ask ladder and return (units_filled, total_cost, avg_price_paid).""" |
| 13 | + filled = 0.0 |
| 14 | + cost = 0.0 |
| 15 | + for price, size in asks: |
| 16 | + if filled >= target_units: |
| 17 | + break |
| 18 | + take = min(float(size), target_units - filled) |
| 19 | + if take <= 0: |
| 20 | + continue |
| 21 | + cost += take * float(price) |
| 22 | + filled += take |
| 23 | + avg_price = cost / filled if filled > 0 else 0.0 |
| 24 | + return filled, cost, avg_price |
| 25 | + |
| 26 | + |
| 27 | +def fee_rate_from_book(book: dict[str, Any]) -> float: |
| 28 | + member = book.get("member") or {} |
| 29 | + return float(book.get("fee_rate", member.get("fee_rate", 0.0)) or 0.0) |
| 30 | + |
| 31 | + |
| 32 | +def simulate_basket_fill(book_data: list[dict[str, Any]], requested_size: float) -> dict[str, Any]: |
| 33 | + """Simulate a mutually-exclusive YES basket at the common executable size. |
| 34 | +
|
| 35 | + A basket only pays out for the minimum size completed across all legs. If |
| 36 | + one leg has 2 units of ask depth and the requested basket is 10 units, the |
| 37 | + executable basket is 2 units, not 10. |
| 38 | + """ |
| 39 | + requested_size = float(requested_size) |
| 40 | + requested_fills: list[float] = [] |
| 41 | + for book in book_data: |
| 42 | + filled, _, _ = simulate_buy_cost(book.get("asks") or [], requested_size) |
| 43 | + requested_fills.append(filled) |
| 44 | + |
| 45 | + effective_size = min(requested_fills) if requested_fills else 0.0 |
| 46 | + effective_size = max(0.0, min(effective_size, requested_size)) |
| 47 | + |
| 48 | + total_cost = 0.0 |
| 49 | + total_fee = 0.0 |
| 50 | + per_member: list[dict[str, Any]] = [] |
| 51 | + if effective_size > EPSILON: |
| 52 | + for book in book_data: |
| 53 | + filled, cost, avg_px = simulate_buy_cost(book.get("asks") or [], effective_size) |
| 54 | + fee_rate = fee_rate_from_book(book) |
| 55 | + fee = fee_rate * avg_px * (1.0 - avg_px) * filled |
| 56 | + total_cost += cost |
| 57 | + total_fee += fee |
| 58 | + member = book.get("member") or {} |
| 59 | + per_member.append( |
| 60 | + { |
| 61 | + "member": str(member.get("question") or "")[:40], |
| 62 | + "filled": filled, |
| 63 | + "avg_px": avg_px, |
| 64 | + "cost": cost, |
| 65 | + "fee": fee, |
| 66 | + } |
| 67 | + ) |
| 68 | + |
| 69 | + edge_dollars = effective_size - total_cost - total_fee |
| 70 | + edge_pct = edge_dollars / effective_size if effective_size > EPSILON else 0.0 |
| 71 | + return { |
| 72 | + "size": requested_size, |
| 73 | + "requested_size": requested_size, |
| 74 | + "effective_size": effective_size, |
| 75 | + "max_fillable_units": effective_size, |
| 76 | + "is_full_size_fillable": effective_size + EPSILON >= requested_size, |
| 77 | + "total_cost": total_cost, |
| 78 | + "total_fee": total_fee, |
| 79 | + "edge_dollars": edge_dollars if effective_size > EPSILON else 0.0, |
| 80 | + "edge_pct": edge_pct, |
| 81 | + "per_member": per_member, |
| 82 | + } |
| 83 | + |
| 84 | + |
| 85 | +def maker_target_price( |
| 86 | + best_bid: float, |
| 87 | + best_ask: float, |
| 88 | + markup: float, |
| 89 | + tick_size: float = DEFAULT_TICK_SIZE, |
| 90 | +) -> float | None: |
| 91 | + """Return a non-crossing maker bid target or None if the spread is too tight.""" |
| 92 | + best_bid = float(best_bid) |
| 93 | + best_ask = float(best_ask) |
| 94 | + markup = float(markup) |
| 95 | + tick_size = float(tick_size) |
| 96 | + if best_ask <= 0 or best_bid < 0 or tick_size <= 0: |
| 97 | + return None |
| 98 | + lower = best_bid + tick_size |
| 99 | + upper = best_ask - tick_size |
| 100 | + if upper + EPSILON < lower: |
| 101 | + return None |
| 102 | + target = max(best_ask - markup, lower) |
| 103 | + target = min(target, upper) |
| 104 | + if target <= 0 or target + EPSILON >= best_ask: |
| 105 | + return None |
| 106 | + return round(target, 6) |
| 107 | + |
| 108 | + |
| 109 | +def zero_maker_stats(n_total_days: int, reason: str) -> dict[str, Any]: |
| 110 | + return { |
| 111 | + "targets": [], |
| 112 | + "n_filled_days": 0, |
| 113 | + "n_total_days": n_total_days, |
| 114 | + "fill_rate": 0.0, |
| 115 | + "avg_edge_given_fill": 0.0, |
| 116 | + "median_edge_given_fill": 0.0, |
| 117 | + "expected_daily_edge_dollars": 0.0, |
| 118 | + "avg_min_leg_sell_size": 0.0, |
| 119 | + "avg_effective_basket_size": 0.0, |
| 120 | + "max_effective_basket_size": 0.0, |
| 121 | + "n_positive_edge_days": 0, |
| 122 | + "n_negative_edge_days": 0, |
| 123 | + "skipped_reason": reason, |
| 124 | + } |
| 125 | + |
| 126 | + |
| 127 | +def capped_expected_daily_edge( |
| 128 | + filled_days: list[dict[str, Any]], |
| 129 | + n_total_days: int, |
| 130 | + basket_size: float, |
| 131 | +) -> dict[str, float]: |
| 132 | + """Compute daily maker PnL capped by observed trade size on the thinnest leg.""" |
| 133 | + if n_total_days <= 0 or not filled_days: |
| 134 | + return { |
| 135 | + "expected_daily_edge_dollars": 0.0, |
| 136 | + "avg_effective_basket_size": 0.0, |
| 137 | + "max_effective_basket_size": 0.0, |
| 138 | + } |
| 139 | + |
| 140 | + basket_size = float(basket_size) |
| 141 | + effective_sizes = [ |
| 142 | + min(basket_size, max(0.0, float(day.get("min_leg_sell_size") or 0.0))) |
| 143 | + for day in filled_days |
| 144 | + ] |
| 145 | + pnl = [ |
| 146 | + float(day.get("edge") or 0.0) * size |
| 147 | + for day, size in zip(filled_days, effective_sizes) |
| 148 | + ] |
| 149 | + return { |
| 150 | + "expected_daily_edge_dollars": sum(pnl) / n_total_days, |
| 151 | + "avg_effective_basket_size": statistics.mean(effective_sizes) if effective_sizes else 0.0, |
| 152 | + "max_effective_basket_size": max(effective_sizes) if effective_sizes else 0.0, |
| 153 | + } |
| 154 | + |
| 155 | + |
| 156 | +def qualifying_trade_size(trades: list[dict[str, Any]], target_price: float) -> float: |
| 157 | + """Return total sell size that could have hit a resting bid at target_price.""" |
| 158 | + target_price = float(target_price) |
| 159 | + total = 0.0 |
| 160 | + for trade in trades: |
| 161 | + try: |
| 162 | + price = float(trade.get("price") or 0.0) |
| 163 | + size = float(trade.get("size") or 0.0) |
| 164 | + except (TypeError, ValueError): |
| 165 | + continue |
| 166 | + if price <= target_price and size > 0: |
| 167 | + total += size |
| 168 | + return total |
0 commit comments