"""Order checks for the TradeStation pack. Run this before orderconfirm. TradeStation's confirm accepts duration and session pairs these rules reject, including a pre-market DAY that sits inactive until 09:30 ET. from validate_order import validate_order, session_bucket result = validate_order(order, asset, session, account) if not result["ok"]: # do not call orderconfirm `rules.json` must sit in the same directory as this file. """ from __future__ import annotations import json import re 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 name in order and order[name] is not None: return order[name] return default def _tif(order): tif = _get(order, "TimeInForce", "timeInForce") or {} if not isinstance(tif, dict): return {} return tif def normalize_duration(value): if value is None: return None text = str(value).strip().upper().replace("_", " ") text = " ".join(text.split()) aliases = rules()["duration_aliases"] return aliases.get(text, text) def canon_order_type(value): if value is None: return None token = str(value).strip() upper = token.upper() if upper == "STOP": return "Stop" for name in rules()["order_types"]: if name.upper() == upper: return name return token def normalize_order(order): """Accept the REST PascalCase body or the MCP camelCase body.""" if not isinstance(order, dict): raise TypeError("order must be a dict") tif = _tif(order) legs = _get(order, "Legs", "legs") or [] norm_legs = [] for leg in legs: if not isinstance(leg, dict): continue norm_legs.append({ "symbol": str(_get(leg, "Symbol", "symbol") or ""), "quantity": _get(leg, "Quantity", "quantity"), "trade_action": _upper(_get(leg, "TradeAction", "tradeAction")), }) return { "account_id": _get(order, "AccountID", "AccountId", "accountID", "accountId"), "symbol": _get(order, "Symbol", "symbol") or (norm_legs[0]["symbol"] if norm_legs else None), "quantity": _get(order, "Quantity", "quantity"), "order_type": canon_order_type(_get(order, "OrderType", "orderType")), "trade_action": _upper(_get(order, "TradeAction", "tradeAction")), "limit_price": _get(order, "LimitPrice", "limitPrice"), "stop_price": _get(order, "StopPrice", "stopPrice"), "duration": normalize_duration(_get(tif, "Duration", "duration")), "expiration": _get(tif, "Expiration", "expiration"), "route": _get(order, "Route", "route"), "order_confirm_id": _get(order, "OrderConfirmID", "orderConfirmID", "orderConfirmId"), "legs": norm_legs, "intent": _get(order, "intent"), "environment": _upper(_get(order, "environment")), } def _upper(value): if value is None: return None return str(value).strip().upper() def underlying_root(symbol): if not symbol: return "" text = str(symbol).strip().upper() if text.startswith("$"): text = text[1:] if text.endswith(".X"): text = text[:-2] root = text.split()[0] root = root.split(".")[0] return root def _as_decimal(value): if value is None or value == "": return None try: return Decimal(str(value).replace(",", "")) except InvalidOperation: return None def _positive(value): number = _as_decimal(value) return number is not None and number > 0 def session_bucket(now, asset, underlying=None): """Name the session from an Eastern clock. Futures outside the equity regular session return futures_unspecified. TradeStation's pages fetched for this pack do not state the daily maintenance break. Do not treat that bucket as permission to send. """ if now.tzinfo is None: now = now.replace(tzinfo=ET) else: now = now.astimezone(ET) asset = _canon_asset(asset, underlying) weekend = now.weekday() >= 5 minutes = now.hour * 60 + now.minute if asset == "futures": if weekend: return "futures_closed" if 9 * 60 + 30 <= minutes < 16 * 60: return "futures_open" return "futures_unspecified" if weekend: return "closed" root = underlying_root(underlying) if underlying else "" select = root in set(rules()["select_equity_option_underlyings"]["symbols"]) index = asset == "index_option" or root in set(rules()["index_option_roots"]) if asset == "option" and index: asset = "index_option" if asset == "index_option": if 7 * 60 + 15 <= minutes < 9 * 60 + 25: return "index_gth" if 9 * 60 + 25 <= minutes < 9 * 60 + 30: return "options_pause" if 9 * 60 + 30 <= minutes < 16 * 60 + 15: return "index_regular" if 16 * 60 + 15 <= minutes < 17 * 60: return "index_curb" return "closed" if asset == "option": if select and 7 * 60 + 30 <= minutes < 9 * 60 + 25: return "options_gth" if select and 9 * 60 + 25 <= minutes < 9 * 60 + 30: return "options_pause" if 9 * 60 + 30 <= minutes < 16 * 60: return "regular" if select and 16 * 60 <= minutes < 16 * 60 + 15: return "options_curb" if 16 * 60 <= minutes < 16 * 60 + 17: return "options_late" return "closed" if 6 * 60 <= minutes < 9 * 60 + 30: return "premarket" if 9 * 60 + 30 <= minutes < 16 * 60: return "regular" if 16 * 60 <= minutes < 20 * 60: return "postmarket" return "closed" def _canon_asset(asset, symbol=None): key = str(asset or "").strip().lower().replace("-", "_").replace(" ", "_") aliases = { "stock": "stock", "equity": "stock", "equities": "stock", "option": "option", "equity_option": "option", "options": "option", "index_option": "index_option", "index": "index_option", "future": "futures", "futures": "futures", "combo": "option", } asset_name = aliases.get(key, key) root = underlying_root(symbol) if asset_name == "option" and root in set(rules()["index_option_roots"]): return "index_option" return asset_name def _duration_code(entry): if isinstance(entry, str): return entry return entry.get("code") def _find_allow(asset, session): for rule in rules()["allow"]: if rule["asset"] == asset and session in rule["sessions"]: return rule return None def _gap_reason(asset, session, duration, order_type, select): for gap in rules()["gaps"]: if gap.get("asset") not in (None, asset): continue if "sessions" in gap and session not in gap["sessions"]: continue if "durations" in gap and duration not in gap["durations"]: continue if "order_types" in gap and order_type not in gap["order_types"]: continue if gap.get("requires_select_underlying") is False and select: continue if gap.get("requires_select_underlying") is True and not select: continue return gap["reason"] return None def _parse_expiration(value, now): if not value: return None, "GTD and GDP need TimeInForce.Expiration as RFC3339. Only the date portion is used." text = str(value).strip() try: parsed = datetime.strptime(text[:10], "%Y-%m-%d") except ValueError: return None, "Expiration must be RFC3339. Example: 2026-10-23T00:00:00Z." expiry = parsed.date() if now.tzinfo is None: today = now.date() else: today = now.astimezone(ET).date() max_days = int(rules()["limits"]["gtc_max_calendar_days"]) if expiry < today: return expiry, "Expiration date is already past." if (expiry - today).days > max_days: return expiry, f"GTD and GTC last at most {max_days} calendar days." return expiry, None def position_qty(account, symbol): """Signed share or contract count for one symbol. Long is positive.""" if not account: return None positions = account.get("positions") or account.get("Positions") or [] root = str(symbol or "") total = Decimal(0) found = False for row in positions: row_symbol = str(_get(row, "Symbol", "symbol") or "") if row_symbol.upper() != str(root).upper(): continue found = True qty = _as_decimal(_get(row, "Quantity", "quantity")) or Decimal(0) side = _upper(_get(row, "LongShort", "longShort")) if side == "SHORT": qty = -abs(qty) elif side == "LONG": qty = abs(qty) total += qty if not found: return None return total def trade_action(intent, asset, position_qty_value, order_qty): """Map open/close long/short onto the API trade action. Read positions first. A close larger than the position is not ok: the extra size opens the other side. """ asset_name = _canon_asset(asset) intent_key = str(intent or "").strip().lower() table = rules()["trade_actions"].get(asset_name) reasons = [] warnings = [] if table is None or intent_key not in table: return {"ok": False, "action": None, "reasons": [f"Unknown intent {intent!r} for {asset_name}."], "warnings": []} action = table[intent_key] qty = _as_decimal(order_qty) held = position_qty_value if isinstance(position_qty_value, Decimal) else _as_decimal(position_qty_value) if intent_key.startswith("close"): if held is None: warnings.append("No position was passed. A close sent against a flat symbol opens a new position.") elif intent_key == "close_long" and held <= 0: reasons.append("Close long requested, and the position is not long. Read positions again before choosing SELL or SELLTOCLOSE.") elif intent_key == "close_short" and held >= 0: reasons.append("Close short requested, and the position is not short.") elif qty is not None and abs(held) and qty > abs(held): reasons.append("Order quantity is larger than the open position. The extra size opens the other side. Lower the quantity or split the order.") if intent_key == "open_short" and asset_name == "stock": warnings.append("SELLSHORT opens a short. It does not close a long. Check IsStockLocateEligible and IsHardToBorrow first.") return {"ok": not reasons, "action": action, "reasons": reasons, "warnings": warnings} def _account_type(account): return _upper(_get(account, "AccountType", "accountType")) def _account_status(account): return str(_get(account, "Status", "status") or "") def _option_level(account): detail = account.get("AccountDetail") or account.get("accountDetail") or {} level = _get(detail, "OptionApprovalLevel", "optionApprovalLevel") if level is None: level = _get(account, "OptionApprovalLevel", "optionApprovalLevel") try: return int(level) if level is not None else None except (TypeError, ValueError): return None def _check_account(norm, asset, account): reasons = [] warnings = [] if account is None: warnings.append("No account was passed. Account-type, option-level, and position checks did not run.") return reasons, warnings status = _account_status(account) kind = _account_type(account) if status in rules()["blocked_statuses"]: reasons.append(f"Account status {status} does not allow a new order.") action = norm["trade_action"] intent = (norm["intent"] or "").lower() closing = intent.startswith("close") or action in {"SELL", "BUYTOCOVER", "SELLTOCLOSE", "BUYTOCLOSE"} if asset == "futures" and action == "BUY" and intent == "close_short": closing = True if status in rules()["closing_only_statuses"] and not closing: reasons.append(f"Account status {status} allows closing transactions only.") if asset == "futures" and kind and kind != "FUTURES": reasons.append("Futures orders belong on the Futures account, not Cash or Margin.") if asset in {"stock", "option", "index_option"} and kind == "FUTURES": reasons.append("This is a Futures account. Equities and options use the Cash or Margin account.") if asset == "stock" and kind == "CASH" and (action == "SELLSHORT" or intent == "open_short"): reasons.append("A Cash account cannot SELLSHORT.") if asset in {"option", "index_option"}: level = _option_level(account) if level == 0: reasons.append("OptionApprovalLevel 0 does not allow options.") elif level is None: warnings.append("OptionApprovalLevel was not on the account object.") else: legs = norm["legs"] option_legs = [leg for leg in legs if " " in leg["symbol"]] naked = action == "SELLTOOPEN" or any(leg["trade_action"] == "SELLTOOPEN" for leg in legs) covered = False if naked and asset == "option": stock_legs = [leg for leg in legs if leg["symbol"] and " " not in leg["symbol"]] held = position_qty(account, underlying_root(norm["symbol"])) contracts = _as_decimal(norm["quantity"]) or Decimal(1) if stock_legs: covered = True if held is not None and held >= contracts * 100: covered = True if naked and not covered and len(option_legs) < 2: need = 5 if asset == "index_option" else 4 if level < need: reasons.append(f"A naked short option needs OptionApprovalLevel {need}. This account is level {level}. Covered calls need the long stock in the order or in positions.") if len(option_legs) >= 2 and level < 3: reasons.append(f"Option spreads need OptionApprovalLevel 3 or higher. This account is level {level}.") return reasons, warnings def _closes_long(asset, action): return (asset == "stock" and action == "SELL") or (asset in {"option", "index_option"} and action == "SELLTOCLOSE") def _closes_short(asset, action): return (asset == "stock" and action == "BUYTOCOVER") or (asset in {"option", "index_option"} and action == "BUYTOCLOSE") def _check_action(norm, asset, account): reasons = [] warnings = [] held = position_qty(account, norm["symbol"]) if account else None if norm["intent"]: picked = trade_action(norm["intent"], asset, held, norm["quantity"]) reasons.extend(picked["reasons"]) warnings.extend(picked["warnings"]) if picked["action"] and norm["trade_action"] and norm["trade_action"] != picked["action"]: reasons.append(f"TradeAction {norm['trade_action']} does not match intent {norm['intent']}, which maps to {picked['action']}.") elif picked["action"] and not norm["trade_action"]: norm["trade_action"] = picked["action"] elif account is not None and _closes_long(asset, norm["trade_action"]): qty = _as_decimal(norm["quantity"]) if held is None or held <= 0: reasons.append("No long position to close. Confirm accepts a SELL or SELLTOCLOSE anyway. Only close what exists.") elif qty is not None and qty > held: reasons.append("Close quantity is larger than the long position. Confirm accepts the extra size. Never close more than you hold.") elif account is not None and _closes_short(asset, norm["trade_action"]): qty = _as_decimal(norm["quantity"]) if held is None or held >= 0: reasons.append("No short position to cover. Confirm accepts a BUYTOCOVER or BUYTOCLOSE anyway. Only close what exists.") elif qty is not None and qty > abs(held): reasons.append("Cover quantity is larger than the short position. Confirm accepts the extra size. Never close more than you hold.") elif asset == "futures" and norm["trade_action"] == "SELL" and account is not None: qty = _as_decimal(norm["quantity"]) if held is not None and held > 0 and qty is not None and qty > held: reasons.append("Order quantity is larger than the long position. The extra size opens a short. Lower the quantity or split the order.") elif held is None or held <= 0: warnings.append("Futures SELL opens a short when the position is not long. Pass intent so an open and a close stay distinct.") elif norm["trade_action"] == "SELLSHORT" and held is not None and held > 0: warnings.append("A long is already open. SELLSHORT does not close it. Send SELL to close the long.") if asset == "futures" and norm["trade_action"] in {"SELLSHORT", "BUYTOCOVER"}: reasons.append("Futures use BUY and SELL. BUYTOCOVER and SELLSHORT are equity actions.") if asset == "stock" and norm["trade_action"] in {"BUYTOOPEN", "SELLTOCLOSE", "SELLTOOPEN", "BUYTOCLOSE"} and not norm["legs"]: reasons.append("Single-stock orders use BUY, SELL, SELLSHORT, or BUYTOCOVER.") if len(norm["legs"]) > 1: for leg in norm["legs"]: if leg["symbol"] and " " not in leg["symbol"] and leg["trade_action"] in {"BUY", "SELL", "SELLSHORT", "BUYTOCOVER"}: warnings.append("The spec's covered-stock example sends the stock leg as BuyToOpen. If confirm says Invalid trade action, resend that leg as BUYTOOPEN or SELLTOOPEN.") break return reasons, warnings def _known_tick(asset_name, symbol): """Stock 0.01 and ES 0.25 are known. Any other product stays unknown.""" ticks = rules().get("ticks") or {} if asset_name == "stock": return _as_decimal(ticks.get("stock")) if asset_name != "futures": return None text = str(symbol or "").strip().upper() for row in ticks.get("futures") or []: pattern = row.get("pattern") or "" if pattern and re.fullmatch(pattern, text): return _as_decimal(row.get("increment")) return None def confirm_price_matches(sent, echoed): """Stop when confirm's price is not the price that was sent. A buy limit of 728.005 was re-priced to 728.01. 1.003 came back as 1.00. """ sent_n = _as_decimal(sent) echoed_n = _as_decimal(echoed) if sent_n is None or echoed_n is None: return {"ok": False, "reasons": ["Could not read both prices. Stop. Do not place on a confirm you cannot compare."]} if sent_n != echoed_n: return {"ok": False, "reasons": [f"summaryMessage price {echoed} differs from the price sent {sent}. Do not place."]} return {"ok": True, "reasons": []} def futures_symbol_kind(symbol): """continuous_root cannot be ordered. individual and continuous_individual can.""" text = str(symbol or "").strip().upper() if not text: return "missing" if "=" in text: return "custom_continuous" if text.startswith("@"): body = text[1:].split(".")[0] if any(ch.isdigit() for ch in body): return "continuous_individual" return "continuous_root" return "individual" def validate_order(order, asset, session, account=None, now=None): """Return ok, reasons, and warnings. Do not send when ok is false.""" data = rules() reasons = [] warnings = [] if now is None: now = datetime.now(ET) elif now.tzinfo is None: now = now.replace(tzinfo=ET) try: norm = normalize_order(order) except TypeError as exc: return {"ok": False, "reasons": [str(exc)], "warnings": [], "normalized": None} symbol = norm["symbol"] asset_name = _canon_asset(asset, symbol) if str(asset).lower() in {"option", "equity_option"} and asset_name == "index_option": warnings.append("Underlying is an index root. Checked as index_option, not as an equity option.") select = underlying_root(symbol) in set(data["select_equity_option_underlyings"]["symbols"]) order_type = norm["order_type"] duration = norm["duration"] if order_type in data["rejected_order_types"]: reasons.append(data["rejected_order_types"][order_type]) elif order_type not in data["order_types"]: reasons.append("OrderType must be Market, Limit, StopMarket, or StopLimit.") if not symbol: reasons.append("Symbol is required.") elif str(symbol) != str(symbol).upper(): reasons.append("Symbol must be uppercase.") if not _positive(norm["quantity"]) and not norm["legs"]: reasons.append("Quantity must be a positive number, sent as a string.") if order_type in {"Limit", "StopLimit"}: # A credit spread is a negative limit. Zero is not a price. limit_px = _as_decimal(norm["limit_price"]) if norm["limit_price"] in (None, ""): reasons.append(f"{order_type} needs LimitPrice as a string. A credit is a negative string.") elif limit_px is None: reasons.append("LimitPrice must be a numeric string.") elif limit_px == 0: reasons.append("LimitPrice cannot be 0.") if order_type in {"StopMarket", "StopLimit"} and not _positive(norm["stop_price"]): reasons.append(f"{order_type} needs StopPrice as a string.") tick = _known_tick(asset_name, symbol) if tick is not None: for label, raw in (("LimitPrice", norm["limit_price"]), ("StopPrice", norm["stop_price"])): if raw in (None, ""): continue number = _as_decimal(raw) if number is not None and number % tick != 0: reasons.append( f"{label} {raw} is off the {format(tick, 'f')} tick. Confirm may reprice it. Send a valid tick, and stop if summaryMessage shows a different price." ) elif asset_name in {"option", "index_option", "futures"} and (norm["limit_price"] not in (None, "") or norm["stop_price"] not in (None, "")): warnings.append("Tick for this product is not in the rules. Do not guess. Compare summaryMessage with the price you sent, and stop if they differ.") if duration in {"GTD", "GDP"}: _, exp_reason = _parse_expiration(norm["expiration"], now) if exp_reason: reasons.append(exp_reason) if norm["order_confirm_id"] and len(str(norm["order_confirm_id"])) > data["limits"]["order_confirm_id_max"]: reasons.append("OrderConfirmID is 1 to 22 characters on the REST path.") if norm["route"] and str(norm["route"]) not in {str(norm["route"]).upper(), "Intelligent"}: warnings.append("The Route description says uppercase, and the spec examples send Intelligent. Use a route id from GET /v3/orderexecution/routes.") if asset_name == "futures": kind = futures_symbol_kind(symbol) if kind == "continuous_root": reasons.append("A continuous root such as @ES is for charts. It cannot be used to place an order. Send the individual contract.") elif kind == "custom_continuous": warnings.append("Custom continuous symbols are a chart setting. Confirm the tradable contract before sending.") allow = _find_allow(asset_name, session) if allow is None: gap = _gap_reason(asset_name, session, duration, order_type, select) if gap: reasons.append(gap) elif asset_name == "index_option": reasons.append("Index option hours outside the regular session are on the extended-hours article. Plus durations are not listed for index options. Do not send until a SIM confirm is recorded.") else: reasons.append(f"No rule for {asset_name} in session {session}.") else: if allow.get("warning"): warnings.append(allow["warning"]) if order_type and order_type in data["order_types"] and order_type not in allow.get("order_types", []): reasons.append(allow.get("order_type_reason") or f"{order_type} is not allowed for {asset_name} in {session}.") matched = None for entry in allow.get("durations", []): if _duration_code(entry) == duration: matched = entry break if duration and matched is None: gap = _gap_reason(asset_name, session, duration, order_type, select) reasons.append(gap or f"Duration {duration} is not allowed for {asset_name} in {session}.") elif isinstance(matched, dict) and matched.get("requires_select_underlying") and not select: gap = _gap_reason(asset_name, session, duration, order_type, False) reasons.append(gap or "This plus duration is only valid for a select equity option underlying.") elif isinstance(matched, dict) and matched.get("order_types") and order_type not in matched["order_types"]: reasons.append(f"{duration} with {order_type} is not allowed in {session}.") if duration and duration in (allow.get("warnings") or {}): warnings.append(allow["warnings"][duration]) if norm["environment"] and norm["environment"] not in {"SIM", "LIVE"}: reasons.append("MCP place-order environment must be SIM or LIVE, and it must match get-trading-environment.") acct_reasons, acct_warnings = _check_account(norm, asset_name, account) act_reasons, act_warnings = _check_action(norm, asset_name, account) reasons.extend(acct_reasons) reasons.extend(act_reasons) warnings.extend(acct_warnings) warnings.extend(act_warnings) # De-duplicate, keep order. 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": None if allow is None else allow["id"], "pack_version": data["version"], } def suggest_duration(want, asset): asset_name = _canon_asset(asset) key = {"index_option": "option"}.get(asset_name, asset_name) for row in rules()["when_to_use"]: if row["id"] == want or row["want"].lower() == str(want).lower(): return row.get(key) return None def status_bucket(code): data = rules()["status"] token = _upper(code) if token in data["filled"]: return "filled" if token in data["canceled"]: return "canceled" if token in data["rejected"]: return "rejected" if token in data["open"]: return "open" return "unknown" def is_partial(code): token = _upper(code) if token == "FPR": return "alive" if token == "FLP": return "terminal" return None def fill_report(order): """Average fill on the order, and per-leg execution price and quantity.""" if not isinstance(order, dict): return {"filled_price": None, "legs": []} legs = [] for leg in _get(order, "Legs", "legs") or []: legs.append({ "symbol": _get(leg, "Symbol", "symbol"), "exec_quantity": _get(leg, "ExecQuantity", "execQuantity"), "quantity_remaining": _get(leg, "QuantityRemaining", "quantityRemaining"), "execution_price": _get(leg, "ExecutionPrice", "executionPrice"), }) return { "filled_price": _get(order, "FilledPrice", "filledPrice"), "status": _get(order, "Status", "status"), "bucket": status_bucket(_get(order, "Status", "status")), "partial": is_partial(_get(order, "Status", "status")), "legs": legs, } def reconcile(log_rows, orders, positions=None): """Match today's broker orders to the agent log. There is no client-tag field on the v3 order. OrderID is the key. OrderConfirmID is the REST dedupe key on the way in. It is not documented on the order record, so a crash before the log line shows up as an order whose id is absent from the log. """ log_rows = log_rows or [] orders = orders or [] logged_ids = {str(row.get("order_id") or row.get("OrderID") or "") for row in log_rows} logged_ids.discard("") logged_confirms = {str(row.get("order_confirm_id") or row.get("OrderConfirmID") or "") for row in log_rows} logged_confirms.discard("") seen_ids = set() unlogged = [] for order in orders: oid = str(_get(order, "OrderID", "orderID", "orderId") or "") confirm = str(_get(order, "OrderConfirmID", "orderConfirmID") or "") if oid: seen_ids.add(oid) if oid in logged_ids or (confirm and confirm in logged_confirms): continue unlogged.append({ "order_id": oid, "symbol": _get(order, "Symbol", "symbol") or (_get((order.get("Legs") or [{}])[0], "Symbol", "symbol")), "status": _get(order, "Status", "status"), }) missing = [] for row in log_rows: oid = str(row.get("order_id") or row.get("OrderID") or "") if oid and oid not in seen_ids: missing.append({"order_id": oid, "symbol": row.get("symbol")}) return { "unlogged_orders": unlogged, "logged_but_not_in_snapshot": missing, "positions": positions or [], } def quote_trust(quote): """Reject a quote that is delayed, halted, restricted, or not entitled.""" reasons = [] warnings = [] if not isinstance(quote, dict): return {"ok": False, "reasons": ["Quote was not an object."], "warnings": []} errors = quote.get("Errors") or quote.get("errors") or [] if isinstance(errors, dict): errors = [errors] for err in errors: text = str(err.get("Error") or err.get("error") or err) if "NOTENTITLED" in text.upper().replace(" ", ""): reasons.append("Quote error NotEntitled. Do not price an order from this quote.") else: reasons.append(f"Quote error: {text}") flags = quote.get("MarketFlags") or quote.get("marketFlags") or {} delayed = flags.get("IsDelayed", flags.get("isDelayed")) halted = flags.get("IsHalted", flags.get("isHalted")) hard = flags.get("IsHardToBorrow", flags.get("isHardToBorrow")) if delayed is True or str(delayed).lower() == "true": reasons.append("IsDelayed is true. Do not price a limit or a receipt from this quote.") if halted is True or str(halted).lower() == "true": reasons.append("IsHalted is true. Do not send.") restrictions = quote.get("Restrictions") or quote.get("restrictions") or [] if restrictions: reasons.append("Restrictions is not empty: " + ", ".join(str(item) for item in restrictions) + ". Do not send until you know what they mean.") if hard is True or str(hard).lower() == "true": warnings.append("IsHardToBorrow is true. Do not SELLSHORT without a locate check.") if not quote.get("Bid") or not quote.get("Ask"): warnings.append("Bid or Ask is missing. Pull another quote before using a price.") return {"ok": not reasons, "reasons": reasons, "warnings": warnings} _DISPLAY_UNIT = { 1: Decimal("1"), 2: Decimal("0.1"), 3: Decimal("0.01"), 4: Decimal("0.001"), 5: Decimal("0.0001"), 6: Decimal("0.00001"), 23: Decimal("0.000001"), } def tick_from_fields(fields): """Price increment and point value from symbol-details or quote fields. Prefer PriceFormat.Increment when IncrementStyle is Simple. MinMove is a multiplier on the display type, not the increment itself. """ if not isinstance(fields, dict): return {"ok": False, "reasons": ["No quote or symbol-details object."]} price_format = fields.get("PriceFormat") or fields.get("priceFormat") or {} reasons = [] increment = None if price_format: style = str(price_format.get("IncrementStyle") or "") raw_inc = price_format.get("Increment") if style and style != "Simple": reasons.append(f"IncrementStyle is {style}, not Simple. Do not guess the tick.") elif raw_inc is not None: increment = _as_decimal(raw_inc) if increment is None or increment <= 0: reasons.append("PriceFormat.Increment is not a positive number.") increment = None point_value = _as_decimal(price_format.get("PointValue") if price_format else None) if point_value is None: point_value = _as_decimal(fields.get("PointValue") if "PointValue" in fields else fields.get("pointValue")) min_move = fields.get("MinMove", fields.get("minMove")) display = fields.get("DisplayType", fields.get("displayType")) if increment is None and min_move is not None and not reasons: move = _as_decimal(min_move) if move is not None and move < 0: reasons.append("MinMove is negative, so the tick depends on price. Use PriceFormat from symbol details instead of guessing.") elif display is None: reasons.append("MinMove without DisplayType or PriceFormat is not a tick size. Fetch symbol details.") else: try: unit = _DISPLAY_UNIT.get(int(display)) except (TypeError, ValueError): unit = None if unit is None: reasons.append("DisplayType is a fraction code. Use PriceFormat.Increment for the tick.") elif move is not None: increment = unit * move if increment is None and not reasons: reasons.append("No PriceFormat.Increment, and MinMove could not be applied.") tick_value = None if increment is not None and point_value is not None: tick_value = increment * point_value return { "ok": increment is not None and not reasons, "increment": None if increment is None else format(increment, "f"), "point_value": None if point_value is None else format(point_value, "f"), "tick_value": None if tick_value is None else format(tick_value, "f"), "reasons": 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 implied_multiplier(estimated_cost, price, quantity): cost = _as_decimal(estimated_cost) px = _as_decimal(price) qty = _as_decimal(quantity) if None in (cost, px, qty) or px == 0 or qty == 0: return None return format(cost / (px * qty), "f") def worked_examples(): """Spec examples, plus the SIM observations that the spec does not state.""" future = tick_from_fields({ "PriceFormat": {"Format": "Fraction", "Fraction": "8", "IncrementStyle": "Simple", "Increment": "0.25", "PointValue": "50"}, }) stock = tick_from_fields({ "PriceFormat": {"Format": "Decimal", "Decimals": "2", "IncrementStyle": "Simple", "Increment": "0.01", "PointValue": "1"}, }) es_quote = tick_from_fields({"MinMove": "25", "PointValue": "50"}) return { "spec_future": future, "spec_stock": stock, "sim_es_min_move": es_quote, "sim_option_multiplier": implied_multiplier("100", "1.00", "1"), } 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(): sample = worked_examples() future_ok = sample["spec_future"]["ok"] and sample["spec_future"]["tick_value"] == "12.50" stock_ok = sample["spec_stock"]["ok"] and sample["spec_stock"]["increment"] == "0.01" es_blocked = sample["sim_es_min_move"]["ok"] is False if not (future_ok and stock_ok and es_blocked): raise SystemExit(f"self-check failed {sample}") print(json.dumps({"pack_version": rules()["version"], "self_check": "ok"}, indent=2)) if __name__ == "__main__": main()