"""Order checks for the Alpaca pack. Run this before POST /v2/orders. Alpaca has no dry-run confirm. The docs disagree on two sends this module rejects: gtc with extended_hours, and a fractional or notional order outside the regular session. from validate_order import validate_order, session_bucket result = validate_order(order, asset, session, account) if not result["ok"]: # do not place `rules.json` must sit in the same directory as this file. """ from __future__ import annotations import json from datetime import datetime from decimal import Decimal, InvalidOperation from pathlib import Path from zoneinfo import ZoneInfo ET = ZoneInfo("America/New_York") RULES_PATH = Path(__file__).with_name("rules.json") _RULES = None def load_rules(path=None): global _RULES raw = json.loads(Path(path or RULES_PATH).read_text()) _RULES = raw return raw def rules(): return _RULES if _RULES is not None else load_rules() def _get(order, *names, default=None): for name in names: if isinstance(order, dict) and name in order and order[name] is not None: return order[name] return default def _as_decimal(value): if value is None or value == "": return None try: return Decimal(str(value).replace(",", "")) except InvalidOperation: return None def _places(value): number = _as_decimal(value) if number is None: return None exponent = number.as_tuple().exponent if not isinstance(exponent, int) or exponent >= 0: return 0 return -exponent def _bool(value): if isinstance(value, bool): return value if value is None: return None text = str(value).strip().lower() if text in {"true", "1", "yes"}: return True if text in {"false", "0", "no"}: return False return None def _lower(value): if value is None: return None return str(value).strip().lower() def _canon_asset(asset): if isinstance(asset, dict): raw = _get(asset, "class", "asset_class", "asset", default="") else: raw = asset or "" token = str(raw).strip().lower().replace("-", "_").replace(" ", "_") if token in {"stock", "etf", "us_equity", "equity"}: return "stock" if token in {"option", "us_option", "options"}: return "option" if token in {"crypto", "crypto_open"}: return "crypto" if token in {"future", "futures", "us_future"}: return "futures" return token def _asset_info(asset): return asset if isinstance(asset, dict) else {} def _flag(name, *sources): for source in sources: if isinstance(source, dict) and name in source and source[name] is not None: return source[name] return None def _minutes(clock): hour, minute = clock.split(":") return int(hour) * 60 + int(minute) def session_bucket(now, asset): """Name the session from an Eastern clock. Crypto ignores the equity clock and returns crypto_open, including weekends. Friday 20:00 ET through Sunday 20:00 ET is closed for stocks and options. """ if now.tzinfo is None: now = now.replace(tzinfo=ET) else: now = now.astimezone(ET) if _canon_asset(asset) == "crypto": return "crypto_open" data = rules()["sessions"] pre = _minutes(data["premarket_start"]) regular = _minutes(data["regular_start"]) post = _minutes(data["postmarket_start"]) overnight = _minutes(data["overnight_start"]) minutes = now.hour * 60 + now.minute weekday = now.weekday() if weekday == 5: return "closed" if weekday == 6 and minutes < overnight: return "closed" if weekday == 4 and minutes >= overnight: return "closed" if minutes < pre or minutes >= overnight: return "overnight" if minutes < regular: return "premarket" if minutes < post: return "regular" return "postmarket" def _host(value): text = str(value or "").strip().rstrip("/").lower() if "://" in text: text = text.split("://", 1)[1] return text.split("/", 1)[0] def normalize_order(order): if not isinstance(order, dict): raise TypeError("order must be a dict") legs = _get(order, "legs", default=[]) or [] norm_legs = [] for leg in legs: if not isinstance(leg, dict): continue norm_legs.append({ "symbol": _get(leg, "symbol"), "side": _lower(_get(leg, "side")), "position_intent": _lower(_get(leg, "position_intent")), "qty": _get(leg, "qty", "ratio_qty"), }) extended = _bool(_get(order, "extended_hours")) return { "symbol": None if _get(order, "symbol") is None else str(_get(order, "symbol")).strip(), "side": _lower(_get(order, "side")), "type": _lower(_get(order, "type", "order_type")), "time_in_force": _lower(_get(order, "time_in_force")), "qty": _get(order, "qty", "quantity"), "notional": _get(order, "notional"), "limit_price": _get(order, "limit_price"), "stop_price": _get(order, "stop_price"), "trail": _get(order, "trail"), "trail_price": _get(order, "trail_price"), "trail_percent": _get(order, "trail_percent"), "extended_hours": extended, "client_order_id": _get(order, "client_order_id"), "order_class": _lower(_get(order, "order_class")) or "simple", "position_intent": _lower(_get(order, "position_intent")), "intent": _lower(_get(order, "intent")), "legs": norm_legs, "take_profit": _get(order, "take_profit") or {}, "stop_loss": _get(order, "stop_loss") or {}, "base_url": _get(order, "base_url", "APCA_API_BASE_URL"), "environment": _lower(_get(order, "environment")), "alpaca_paper_trade": _bool(_get(order, "alpaca_paper_trade", "ALPACA_PAPER_TRADE")), "live_trading": _bool(_get(order, "live_trading", "LIVE_TRADING")), "feed": _lower(_get(order, "feed")), "gtc_extended_enabled": _bool(_get(order, "gtc_extended_enabled")), "sales_enabled": _bool(_get(order, "sales_enabled")), "overnight_enabled": _bool(_get(order, "overnight_enabled")), "sip_entitled": _bool(_get(order, "sip_entitled")), } def _subpenny_reason(field, value): number = _as_decimal(value) places = _places(value) if number is None or places is None: return f"{field} is not a number." if number < 0: return f"{field} is negative. Stop and ask before sending a credit." data = rules()["subpenny"] cutoff = Decimal(data["at_or_above_dollars"]) allowed = data["places_at_or_above"] if number >= cutoff else data["places_below"] if places > allowed: return ( f"invalid {field} {value}. sub-penny increment does not fulfill " f"minimum pricing criteria ({data['code']})." ) return None def _check_host(norm, account): reasons = [] data = rules()["hosts"] paper = _host(data["paper"]) live = _host(data["live"]) host = _host(norm["base_url"]) if norm["base_url"] else "" env = norm["environment"] paper_flag = norm["alpaca_paper_trade"] if paper_flag is None and isinstance(account, dict): paper_flag = _bool(_flag("alpaca_paper_trade", account)) live_trading = norm["live_trading"] is True or ( isinstance(account, dict) and _bool(_flag("live_trading", account)) is True ) named_paper = host == paper or env == "paper" or paper_flag is True named_live = host == live or env == "live" or paper_flag is False if host and host not in {paper, live}: reasons.append(f"Host {host} is not the paper or live Trading API host.") if named_paper and named_live: reasons.append("Paper and live are both set. Say one host.") if not named_paper and not named_live: reasons.append(f"Say the host. Paper is {data['paper']}.") if named_live and not live_trading: reasons.append("Live host or ALPACA_PAPER_TRADE false, and live trading is not on. Do not send.") if named_paper and live_trading: reasons.append("Live trading is on, but the host is still paper.") return reasons def _check_account(account): reasons = [] warnings = [] if not isinstance(account, dict): return ["Read GET /v2/account before you send."], [] status = account.get("status") if status is None: reasons.append("Read account status. Send only when it is ACTIVE.") elif str(status).upper() != "ACTIVE": reasons.append(f"Account status is {status}, not ACTIVE. Do not send.") for flag in ("trading_blocked", "account_blocked", "trade_suspended_by_user"): value = _bool(account.get(flag)) if flag in account else None if value is None: reasons.append(f"Read {flag} on the account.") elif value: reasons.append(f"{flag} is true. Do not send.") if "pattern_day_trader" in account or "daytrade_count" in account: reasons.append( "pattern_day_trader or daytrade_count is still on the account. Stop and ask. Do not enforce a $25,000 gate." ) return reasons, warnings def _position(account, symbol): if not isinstance(account, dict) or "positions" not in account: return None wanted = str(symbol or "").upper() for row in account.get("positions") or []: if str(row.get("symbol") or "").upper() == wanted: return row return {"missing": True} def _check_position(norm, asset_name, account): reasons = [] warnings = [] if asset_name == "futures": return reasons, warnings row = _position(account, norm["symbol"]) intent = norm["intent"] if row is None: warnings.append("Read the position before you send.") if intent in {"close_long", "close_short", "open_short"}: reasons.append("Read the position before this close or short.") return reasons, warnings if row.get("missing"): if intent in {"close_long", "close_short"}: reasons.append("No position to close.") if norm["side"] == "sell" and asset_name == "crypto": reasons.append("Crypto cannot be shorted.") if asset_name == "stock" and norm["side"] == "sell" and intent != "close_long": reasons.extend(_check_short(norm, account)) return reasons, warnings if "side" not in row: reasons.append("Position payload does not tell long from short. Stop. Do not guess a sign on qty.") return reasons, warnings side = _lower(row.get("side")) have = _as_decimal(row.get("qty")) want = _as_decimal(norm["qty"]) if side not in {"long", "short"}: reasons.append("Position side is not long or short. Stop.") return reasons, warnings if intent == "close_long" and side != "long": reasons.append("Intent is close long, and the position is not long.") if intent == "close_short" and side != "short": reasons.append("Intent is close short, and the position is not short.") if intent == "open_short" and side == "long": reasons.append("A long is open. Do not send an open short against it.") if side == "long" and norm["side"] == "sell" and want is not None and have is not None and want > have: reasons.append("Sell quantity is larger than the long. Cut it to the long, or stop.") if side == "short" and norm["side"] == "buy" and want is not None and have is not None and want > have: reasons.append("Buy quantity is larger than the short. Cut it to the short, or stop.") if asset_name == "stock" and ( intent == "open_short" or (norm["side"] == "sell" and side != "long" and intent != "close_long") ): reasons.extend(_check_short(norm, account)) return reasons, warnings def _check_short(norm, account): reasons = [] if _as_decimal(norm["qty"]) is not None and _as_decimal(norm["qty"]) != _as_decimal(norm["qty"]).to_integral_value(): reasons.append("A fractional order cannot open a short.") if norm["notional"] not in (None, ""): reasons.append("A notional order cannot open a short.") info = account if isinstance(account, dict) else {} if _bool(info.get("shorting_enabled")) is not True: reasons.append("shorting_enabled is not true. Do not short.") asset = info.get("asset") if isinstance(info.get("asset"), dict) else info if isinstance(asset, dict): if _bool(asset.get("shortable")) is False: reasons.append("Asset is not shortable.") if "easy_to_borrow" in asset and _bool(asset.get("easy_to_borrow")) is not True: reasons.append("easy_to_borrow is not true. No locate call is in this pack. Stop.") elif "easy_to_borrow" not in asset: reasons.append("Read easy_to_borrow. If it is false, stop.") return reasons def _fractional(norm): qty = _as_decimal(norm["qty"]) if norm["notional"] not in (None, ""): return True return qty is not None and qty != qty.to_integral_value() def validate_order(order, asset, session, account=None, now=None): data = rules() norm = normalize_order(order) asset_name = _canon_asset(asset) info = _asset_info(asset) if session in (None, "", "auto"): if now is None: session = None else: session = session_bucket(now, asset) reasons = [] warnings = [] if asset_name == "futures": reasons.append("Futures are not in this pack. Stop.") if asset_name not in {"stock", "option", "crypto", "futures"}: reasons.append(f"Asset {asset_name or '(blank)'} is not stock, option, or crypto.") if session is None: reasons.append("Name the session from the US Eastern clock.") reasons.extend(_check_host(norm, account)) account_reasons, account_warnings = _check_account(account) reasons.extend(account_reasons) warnings.extend(account_warnings) client_id = norm["client_order_id"] if client_id is None or str(client_id).strip() == "": reasons.append("client_order_id is required. Alpaca's generated id will not match your log.") elif len(str(client_id)) > data["client_order_id_max"]: reasons.append(f"client_order_id is longer than {data['client_order_id_max']} characters.") symbol = norm["symbol"] or "" if not symbol: reasons.append("Symbol is missing.") elif symbol != symbol.upper(): reasons.append("Symbol must be uppercase.") if asset_name == "crypto" and symbol and "/" not in symbol: reasons.append("Crypto symbol needs a slash, for example BTC/USD.") if asset_name == "option" and " " in symbol: reasons.append("Option symbol is OCC. Do not put a space in it.") if norm["side"] not in data["sides"]: reasons.append("side is buy or sell.") if norm["qty"] in (None, "") and norm["notional"] in (None, ""): reasons.append("Pass qty or notional.") if norm["qty"] not in (None, "") and norm["notional"] not in (None, ""): reasons.append("Both qty and notional. That is HTTP 400.") for field in ("qty", "notional"): if norm[field] in (None, ""): continue number = _as_decimal(norm[field]) if number is None or number <= 0: reasons.append(f"{field} must be a positive number.") elif _places(norm[field]) is not None and _places(norm[field]) > data["fractional_decimals"]: reasons.append(f"{field} has more than {data['fractional_decimals']} decimal places.") extended = norm["extended_hours"] is True order_class = norm["order_class"] order_type = norm["type"] tif = norm["time_in_force"] fractional = _fractional(norm) complex_class = order_class in data["complex_classes"] extended_session = session in {"premarket", "postmarket", "overnight"} if norm["feed"] == "sip" and norm["sip_entitled"] is not True and not ( isinstance(account, dict) and _bool(account.get("sip_entitled")) is True ): reasons.append("feed=sip is the full tape. On Basic, pass feed=iex.") elif norm["feed"] == "overnight": reasons.append("feed=overnight returns an error. Overnight history is feed=boats.") elif asset_name == "stock" and not norm["feed"]: warnings.append("No feed on the order. On Basic, pass feed=iex.") if asset_name == "stock" and session == "regular" and extended and order_type == "limit" and tif == "day": warnings.append("extended_hours true on a regular-session day limit can keep working after 16:00.") if asset_name == "stock" and session == "overnight" and tif == "day": warnings.append("A day order placed overnight stays working through the next regular session and cancels at 20:00 ET.") if asset_name == "stock": reasons.extend(_check_stock(norm, info, account, session, extended, fractional, complex_class)) elif asset_name == "option": reasons.extend(_check_option(norm, account, session, extended)) elif asset_name == "crypto": reasons.extend(_check_crypto(norm, info, session, extended)) if _crypto_notional_unknown(norm): warnings.append("Crypto notional was not computed. One order is capped at $200,000.") if asset_name in {"stock", "crypto"}: for field in ("limit_price", "stop_price"): if norm[field] in (None, ""): continue if asset_name == "crypto": increment = _as_decimal(_flag("price_increment", info)) if increment is None: warnings.append("Read price_increment on the crypto asset. Do not assume a penny.") elif not _on_tick(norm[field], increment): reasons.append(f"{field} is off the asset price_increment.") else: reason = _subpenny_reason(field, norm[field]) if reason: reasons.append(reason) if order_type in {"limit", "stop_limit"} and norm["limit_price"] in (None, ""): reasons.append("limit_price is required for this type.") if order_type in {"stop", "stop_limit"} and norm["stop_price"] in (None, ""): reasons.append("stop_price is required for this type.") if order_type == "trailing_stop" and norm["trail"] in (None, "") and norm["trail_price"] in (None, "") and norm["trail_percent"] in (None, ""): reasons.append("A trailing stop needs trail_price or trail_percent.") if asset_name != "futures": pos_reasons, pos_warnings = _check_position(norm, asset_name, account if not info else {**(account or {}), "asset": {**(account or {}).get("asset", {}), **info} if isinstance(account, dict) else info}) reasons.extend(pos_reasons) warnings.extend(pos_warnings) rule = None if asset_name == "stock" and session == "regular": rule = "stock-regular" elif asset_name == "stock" and extended_session: rule = "stock-extended" elif asset_name == "option" and session == "regular": rule = "option-regular" elif asset_name == "crypto": rule = "crypto" reasons = list(dict.fromkeys(reasons)) warnings = list(dict.fromkeys(warnings)) return { "ok": not reasons, "reasons": reasons, "warnings": warnings, "normalized": norm, "asset": asset_name, "session": session, "rule": rule, "pack_version": data["version"], } def _gtc_extended(norm, account): if norm["gtc_extended_enabled"] is True: return True return isinstance(account, dict) and _bool(account.get("gtc_extended_enabled")) is True def _sales(norm, account): if norm["sales_enabled"] is True: return True return isinstance(account, dict) and _bool(account.get("sales_enabled")) is True def _overnight_ok(norm, account): if norm["overnight_enabled"] is True: return True return isinstance(account, dict) and _bool(account.get("overnight_enabled")) is True def _check_stock(norm, info, account, session, extended, fractional, complex_class): data = rules() reasons = [] order_type = norm["type"] tif = norm["time_in_force"] if order_type not in data["stock_types_regular"]: reasons.append("Stock type must be market, limit, stop, stop_limit, or trailing_stop.") return reasons if tif is None: reasons.append("time_in_force is missing.") return reasons if session == "closed": reasons.append("Equity session is closed. A queued next-day order is not today's plan. Do not send.") return reasons if session not in {"regular", "premarket", "postmarket", "overnight"}: reasons.append(f"Session {session} is not a stock session.") return reasons starred = tif in data["starred_tif"] if starred and not _sales(norm, account): reasons.append(f"{tif} is starred. Do not send it unless sales enabled that duration.") if complex_class and extended: reasons.append(f"{norm['order_class']} does not go out with extended_hours true.") if complex_class and session != "regular": reasons.append(f"{norm['order_class']} stays on the regular session.") if norm["order_class"] in {"bracket", "oco", "oto"}: reasons.extend(_complex_geometry(norm)) if norm["order_class"] == "oco" and order_type != "limit": reasons.append("OCO type is limit.") if order_type == "trailing_stop" and session != "regular": reasons.append("A trailing stop does not trigger outside regular hours.") if _bool(info.get("ipo")) is True and order_type != "limit": reasons.append("An IPO is limit-only until the first trade.") if fractional: if tif != "day": reasons.append("Fractional and notional orders use time_in_force day.") if session != "regular": reasons.append( "Fractional or notional outside the regular session. The fractional page disagrees with itself. Do not send." ) if _bool(info.get("fractionable")) is False: reasons.append("requested asset is not fractionable.") if norm["side"] == "sell" and norm["intent"] == "open_short": reasons.append("A fractional order cannot open a short.") if session == "regular": if tif not in data["stock_tif_regular"] and not starred: reasons.append("Regular-session stock duration is day or gtc, unless a starred duration is enabled.") if extended and order_type != "limit": reasons.append("extended_hours is only for a limit.") if extended and tif == "gtc" and not _gtc_extended(norm, account): reasons.append("Do not send gtc with extended_hours unless the operator says that enablement is on.") if complex_class and tif not in {"day", "gtc"}: reasons.append("Bracket, OCO, and OTO use day or gtc.") return reasons if order_type != "limit": reasons.append("Extended hours and overnight are limit only.") if not extended: reasons.append("Set extended_hours true. A day order without it queues for the next trading day.") if tif == "gtc" and not _gtc_extended(norm, account): reasons.append("Send time_in_force day. Do not send gtc with extended_hours unless the operator says that enablement is on.") elif tif != "day" and not (tif == "gtc" and _gtc_extended(norm, account)) and not starred: reasons.append("Extended hours time_in_force is day.") if session == "overnight": if not _overnight_ok(norm, account): reasons.append("Do not send overnight until the operator says this account has overnight turned on.") tradable = _bool(info.get("overnight_tradable")) if tradable is not True: reasons.append("Read overnight_tradable. Do not send overnight unless it is true.") if _bool(info.get("overnight_halted")) is True: reasons.append("overnight_halted is true. Do not send into a halt.") return reasons def _check_option(norm, account, session, extended): data = rules() reasons = [] if session != "regular": reasons.append("Options trade the regular session only.") if extended: reasons.append("extended hours is not supported for options trading.") tif = norm["time_in_force"] if tif not in data["option_tif"]: reasons.append("invalid option time_in_force. Options use day or gtc.") mleg = norm["order_class"] == "mleg" allowed = data["option_mleg_types"] if mleg else data["option_types"] if norm["type"] not in allowed: reasons.append("Option type is not on the list for this order class.") if norm["type"] == "trailing_stop": reasons.append("No trailing stop on options.") if norm["notional"] not in (None, ""): reasons.append("notional orders are not supported for options trading.") qty = _as_decimal(norm["qty"]) if qty is not None and qty != qty.to_integral_value(): reasons.append("qty must be integer for options trading.") intent = norm["position_intent"] if intent not in data["position_intents"]: reasons.append("On an option, send position_intent.") elif norm["side"] == "buy" and intent not in {"buy_to_open", "buy_to_close"}: reasons.append("position intent does not match side buy.") elif norm["side"] == "sell" and intent not in {"sell_to_open", "sell_to_close"}: reasons.append("position intent does not match side sell.") if norm["type"] in {"limit", "stop_limit"} and norm["limit_price"] in (None, ""): reasons.append("limit_price is required for this type.") if norm["type"] in {"stop", "stop_limit"} and norm["stop_price"] in (None, ""): reasons.append("stop_price is required for this type.") if norm["limit_price"] not in (None, "") and _as_decimal(norm["limit_price"]) is not None and _as_decimal(norm["limit_price"]) < 0: reasons.append("Option limit sign for a credit is not in this pack. Stop and ask.") if mleg: if not norm["legs"]: reasons.append("A multi-leg order needs legs.") for index, leg in enumerate(norm["legs"]): if leg["position_intent"] not in data["position_intents"]: reasons.append(f"leg.{index} position intent is required.") level = None if isinstance(account, dict) and account.get("options_approved_level") is not None: level = _as_decimal(account.get("options_approved_level")) if level is not None and level < 3: reasons.append("Spreads need options level 3.") if isinstance(account, dict) and account.get("options_approved_level") is not None: level = _as_decimal(account.get("options_approved_level")) if level is not None and level <= 0: reasons.append("account not eligible to trade options.") return reasons def _leg_price(block, field): if not isinstance(block, dict): return None raw = block.get(field) if raw in (None, ""): return None return _as_decimal(raw) def _complex_geometry(norm): """Bracket and limit OTO: stop is at least $0.01 from the entry, on the exit side. OCO: the base is the take-profit. A sell stop is at least $0.01 below it. A buy stop is at least $0.01 above it. """ if norm["order_class"] == "oco": return _oco_geometry(norm) if norm["order_class"] == "oto": return _oto_geometry(norm) return _bracket_geometry(norm) def _bracket_geometry(norm): """Buy: take_profit > entry > stop. Sell is the reverse. Stop is at least $0.01 from entry.""" reasons = [] take = _leg_price(norm["take_profit"], "limit_price") stop = _leg_price(norm["stop_loss"], "stop_price") if take is None or stop is None: reasons.append("A bracket needs take_profit.limit_price and stop_loss.stop_price.") return reasons entry = _as_decimal(norm["limit_price"]) if norm["limit_price"] not in (None, "") else None side = norm["side"] if side == "buy": ordered = take > entry > stop if entry is not None else take > stop if not ordered: reasons.append("A buy bracket needs take_profit above the entry and the entry above the stop.") elif side == "sell": ordered = take < entry < stop if entry is not None else take < stop if not ordered: reasons.append("A sell bracket needs take_profit below the entry and the entry below the stop.") if entry is not None and abs(entry - stop) < Decimal("0.01"): reasons.append("The stop must be at least $0.01 from the entry.") return reasons def _oco_geometry(norm): """Sell OCO: stop at least $0.01 below the take-profit. Buy OCO is the reverse.""" reasons = [] take = _leg_price(norm["take_profit"], "limit_price") stop = _leg_price(norm["stop_loss"], "stop_price") if take is None or stop is None: reasons.append("An OCO needs take_profit.limit_price and stop_loss.stop_price.") return reasons gap = Decimal("0.01") side = norm["side"] if side == "sell" and take - stop < gap: reasons.append("An OCO sell stop must be at least $0.01 below the take-profit.") elif side == "buy" and stop - take < gap: reasons.append("An OCO buy stop must be at least $0.01 above the take-profit.") return reasons def _oto_geometry(norm): """A limit OTO stop is at least $0.01 from the entry, on the exit side of that entry.""" reasons = [] stop = _leg_price(norm["stop_loss"], "stop_price") if stop is None: return reasons entry = _as_decimal(norm["limit_price"]) if norm["limit_price"] not in (None, "") else None if entry is None: return reasons gap = Decimal("0.01") side = norm["side"] if side == "buy" and entry - stop < gap: reasons.append("The stop must be at least $0.01 below the entry on a buy OTO.") elif side == "sell" and stop - entry < gap: reasons.append("The stop must be at least $0.01 above the entry on a sell OTO.") return reasons def _crypto_notional(norm): cap = Decimal(str(rules()["crypto_notional_cap"])) reasons = [] notional = _as_decimal(norm["notional"]) if norm["notional"] not in (None, "") else None if notional is not None and notional > cap: reasons.append("Crypto notional is over $200,000. One order is capped at $200,000.") return reasons qty = _as_decimal(norm["qty"]) if norm["qty"] not in (None, "") else None limit = _as_decimal(norm["limit_price"]) if norm["limit_price"] not in (None, "") else None if qty is not None and limit is not None and qty * limit > cap: reasons.append("Crypto notional is over $200,000. One order is capped at $200,000.") return reasons def _crypto_notional_unknown(norm): if norm["notional"] not in (None, ""): return _as_decimal(norm["notional"]) is None qty = _as_decimal(norm["qty"]) if norm["qty"] not in (None, "") else None limit = _as_decimal(norm["limit_price"]) if norm["limit_price"] not in (None, "") else None return qty is None or limit is None def _check_crypto(norm, info, session, extended): data = rules() reasons = [] if extended: reasons.append("Crypto does not use extended_hours.") if norm["type"] not in data["crypto_types"]: reasons.append("Crypto type is market, limit, or stop_limit.") else: allowed = data["crypto_tif_by_type"].get(norm["type"], []) if norm["time_in_force"] not in allowed: if norm["type"] == "stop_limit": reasons.append("Crypto stop_limit time in force is gtc only.") else: reasons.append("Crypto market and limit time in force is gtc or ioc.") if norm["order_class"] not in {"simple", None}: reasons.append("Crypto order_class is simple only.") reasons.extend(_crypto_notional(norm)) if norm["intent"] == "open_short" or norm["side"] == "sell" and norm["intent"] == "open_short": reasons.append("Crypto cannot be shorted.") if _bool(info.get("shortable")) is True: reasons.append("Do not treat a crypto asset as shortable.") if session not in {None, "crypto_open", "regular", "premarket", "postmarket", "overnight", "closed"}: reasons.append(f"Session {session} is not a crypto session.") return reasons def _on_tick(price, increment): number = _as_decimal(price) step = _as_decimal(increment) if number is None or step is None or step <= 0: return False return number % step == 0 def validate_replace(patch, original=None): """A replace is a new order id. Reject fields the reference does not list.""" data = rules() reasons = [] if not isinstance(patch, dict): return {"ok": False, "reasons": ["Replace body must be an object."], "warnings": []} forbidden = [name for name in ("symbol", "side", "type", "order_class") if name in patch] if forbidden: reasons.append("Do not send " + ", ".join(forbidden) + " on a replace.") allowed = set(data["replace_fields"]) unknown = [name for name in patch if name not in allowed and name not in {"symbol", "side", "type", "order_class"}] if unknown: reasons.append("Replace field is not in the reference: " + ", ".join(unknown) + ".") if set(patch) <= {"client_order_id"}: reasons.append("client_order_id cannot be the only field you change.") if "notional" in patch: reasons.append("A non-IPO notional order cannot be replaced.") if isinstance(original, dict): if _lower(original.get("order_class")) == "oto": reasons.append("Replacing an OTO is not supported yet. Cancel and resend.") if original.get("notional") not in (None, "") and _bool(original.get("ipo")) is not True: reasons.append("A non-IPO notional order cannot be replaced. Cancel and resend.") old_price = "trail_price" if original.get("trail_price") not in (None, "") else None old_percent = "trail_percent" if original.get("trail_percent") not in (None, "") else None if old_price and patch.get("trail_percent") not in (None, ""): reasons.append("You cannot switch between trail_price and trail_percent.") if old_percent and patch.get("trail_price") not in (None, ""): reasons.append("You cannot switch between trail_price and trail_percent.") return {"ok": not reasons, "reasons": reasons, "warnings": []} def status_bucket(code): data = rules()["status"] token = _lower(code) if token in data["ask"]: return "unknown" if token == "filled": return "filled" if token in {"canceled", "expired"}: return "canceled" if token == "rejected": return "rejected" if token == "replaced": return "replaced" if token in data["open"]: return "open" if token in data["final"]: return "final" return "unknown" def reconcile(log_rows, orders, positions=None): """Match broker orders to the log by client_order_id, then by order id.""" log_rows = log_rows or [] orders = orders or [] logged_client = {str(row.get("client_order_id") or "") for row in log_rows} logged_client.discard("") logged_ids = {str(row.get("order_id") or row.get("id") or "") for row in log_rows} logged_ids.discard("") seen = set() unlogged = [] for order in orders: oid = str(_get(order, "id", "order_id") or "") client = str(_get(order, "client_order_id") or "") if oid: seen.add(oid) if client in logged_client or (oid and oid in logged_ids): continue unlogged.append({ "order_id": oid, "client_order_id": client, "symbol": _get(order, "symbol"), "status": _get(order, "status"), }) missing = [] for row in log_rows: oid = str(row.get("order_id") or row.get("id") or "") if oid and oid not in seen: missing.append({"order_id": oid, "client_order_id": row.get("client_order_id"), "symbol": row.get("symbol")}) return { "unlogged_orders": unlogged, "logged_but_not_in_snapshot": missing, "positions": positions or [], } def quote_trust(quote, sip_entitled=False): reasons = [] warnings = [] if not isinstance(quote, dict): return {"ok": False, "reasons": ["Quote was not an object."], "warnings": []} feed = _lower(quote.get("feed")) entitled = sip_entitled or _bool(quote.get("sip_entitled")) is True if feed == "sip" and not entitled: reasons.append("feed=sip is the full tape. On Basic, pass feed=iex. A paper-only account is entitled to IEX.") elif feed == "overnight": reasons.append("feed=overnight returns an error. Use feed=boats for overnight history.") elif not feed: warnings.append("No feed on the quote. On Basic, pass feed=iex.") if not quote.get("bid") or not quote.get("ask"): warnings.append("Bid or ask is missing. Do not invent them.") return {"ok": not reasons, "reasons": reasons, "warnings": warnings} def staleness(today=None): data = rules() checked = datetime.strptime(data["checked"], "%Y-%m-%d").date() if today is None: today = datetime.now(ET).date() elif isinstance(today, datetime): today = today.date() age = (today - checked).days limit = int(data["staleness_days"]) return {"stale": age > limit, "age_days": age, "checked": data["checked"], "limit_days": limit, "version": data["version"]} def main(): print(json.dumps({"pack_version": rules()["version"], "self_check": "ok"}, indent=2)) if __name__ == "__main__": main()