Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 60 additions & 0 deletions application/broker_reconciliation.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,6 +63,53 @@ def _number(value: object, *, field_name: str) -> float:
raise IBKRReconciliationReadError(f"IBKR reconciliation is missing {field_name}.") from exc


def _order_identity_integer(value: object, *, field_name: str) -> int:
if isinstance(value, bool):
raise IBKRReconciliationReadError(
f"IBKR reconciliation order identity has an invalid {field_name}."
)
try:
return int(str(value).strip(), 10)
except (TypeError, ValueError) as exc:
raise IBKRReconciliationReadError(
f"IBKR reconciliation order identity has an invalid {field_name}."
) from exc


def build_canonical_order_key(
*,
account_id: object,
client_id: object,
order_id: object,
perm_id: object,
) -> str:
"""Build the account-scoped IBKR key used by submit and reconciliation paths."""

account = _text(account_id)
if not account or "|" in account or any(ord(character) < 32 for character in account):
raise IBKRReconciliationReadError(
"IBKR reconciliation order identity has an invalid account id."
)
normalized_order_id = _order_identity_integer(order_id, field_name="order id")
if normalized_order_id <= 0:
normalized_perm_id = _order_identity_integer(perm_id, field_name="perm id")
if normalized_perm_id <= 0:
raise IBKRReconciliationReadError(
"IBKR reconciliation order identity requires a positive perm id for a manual order."
)
return f"ibkr|account={account}|perm_id={normalized_perm_id}"

normalized_client_id = _order_identity_integer(client_id, field_name="client id")
if normalized_client_id < 0:
raise IBKRReconciliationReadError(
"IBKR reconciliation order identity requires a non-negative client id."
)
return (
f"ibkr|account={account}|client_id={normalized_client_id}"
f"|order_id={normalized_order_id}"
)


def normalize_account_ids(account_ids: Iterable[str] | str | None) -> tuple[str, ...]:
if account_ids is None:
return ()
Expand Down Expand Up @@ -165,6 +212,12 @@ def _normalise_open_trade(trade: Any, *, selected_account_ids: tuple[str, ...])
raise IBKRReconciliationReadError("IBKR reconciliation received an open order without a contract.")
return {
"account": _text(account_id),
"order_key": build_canonical_order_key(
account_id=account_id,
client_id=getattr(order, "clientId", None),
order_id=getattr(order, "orderId", None),
perm_id=getattr(order, "permId", None),
),
"contract": _safe_contract_fields(contract),
"perm_id": _text(getattr(order, "permId", "")),
"action": _text(getattr(order, "action", "")).upper(),
Expand Down Expand Up @@ -201,6 +254,12 @@ def _normalise_execution(fill: Any, *, selected_account_ids: tuple[str, ...]) ->
raise IBKRReconciliationReadError("IBKR reconciliation received an incomplete execution record.")
return {
"account": _text(account_id),
"order_key": build_canonical_order_key(
account_id=account_id,
client_id=getattr(execution, "clientId", None),
order_id=getattr(execution, "orderId", None),
perm_id=getattr(execution, "permId", None),
),
"contract": _safe_contract_fields(contract),
"execution_id": _text(getattr(execution, "execId", "")),
"order_id": _text(getattr(execution, "orderId", "")),
Expand Down Expand Up @@ -504,6 +563,7 @@ def matches(key: str, actual_digest: str) -> bool:
"IBKRReconciliationCandidate",
"IBKRReconciliationObservations",
"IBKRReconciliationReadError",
"build_canonical_order_key",
"build_reconciliation_candidate",
"collect_read_only_reconciliation_observations",
"normalize_account_ids",
Expand Down
37 changes: 36 additions & 1 deletion application/execution_service.py
Original file line number Diff line number Diff line change
Expand Up @@ -302,6 +302,25 @@ def _record_order_outcome(
"""
normalized_status = str(status or "").strip()
prefix = "option_orders" if option_order else "orders"
identity_failure = str(order_payload.get("reconciliation_outcome") or "").strip()
broker_order_id = str(order_payload.get("broker_order_id") or "").strip()
order_key = str(order_payload.get("order_key") or "").strip()
identity_required_statuses = (
_FILLED_ORDER_STATUSES
| _PARTIALLY_FILLED_ORDER_STATUSES
| _PENDING_ORDER_STATUSES
)
if identity_failure or (
normalized_status in identity_required_statuses
and not order_key
and not broker_order_id
):
reason = identity_failure or "order_identity_unavailable"
execution_summary[f"{prefix}_skipped"].append({**order_payload, "reason": reason})
execution_summary["skipped_reasons"].append(
f"{reason}:{order_payload.get('symbol')}"
)
return "failed"
if normalized_status in _FILLED_ORDER_STATUSES:
execution_summary[f"{prefix}_filled"].append(order_payload)
return "filled"
Expand All @@ -321,6 +340,17 @@ def _record_order_outcome(
return "failed"


def _report_order_identity(report: object) -> dict[str, str]:
raw_payload = getattr(report, "raw_payload", None)
if not isinstance(raw_payload, Mapping):
return {}
return {
key: str(raw_payload[key]).strip()
for key in ("order_key", "reconciliation_outcome")
if str(raw_payload.get(key) or "").strip()
}


def _normalize_account_ids(account_ids=None) -> tuple[str, ...]:
if account_ids is None:
return ()
Expand Down Expand Up @@ -842,6 +872,7 @@ def _execute_option_order_intents(
**payload,
"status": status,
"broker_order_id": getattr(report, "broker_order_id", None),
**_report_order_identity(report),
}
outcome = _record_order_outcome(
execution_summary,
Expand Down Expand Up @@ -2116,6 +2147,7 @@ def cash_sweep_sale_quantity_to_fund_buy(max_quantity: int, candidate_symbols: t
"quantity": qty,
"status": status,
"broker_order_id": getattr(report, "broker_order_id", None),
**_report_order_identity(report),
}
outcome = _record_order_outcome(execution_summary, order_payload, status=status)
trade_logs.append(translator("market_sell", symbol=symbol, qty=format_quantity(qty)) + f" {status_msg}")
Expand Down Expand Up @@ -2301,6 +2333,7 @@ def cash_sweep_sale_quantity_to_fund_buy(max_quantity: int, candidate_symbols: t
"limit_price": limit_price,
"status": status,
"broker_order_id": getattr(report, "broker_order_id", None),
**_report_order_identity(report),
}
outcome = _record_order_outcome(execution_summary, order_payload, status=status)
trade_logs.append(
Expand Down Expand Up @@ -2337,7 +2370,9 @@ def cash_sweep_sale_quantity_to_fund_buy(max_quantity: int, candidate_symbols: t
(
reason
for reason in execution_summary["skipped_reasons"]
if reason.startswith(("submit_failed:", "option_submit_failed:"))
if reason.startswith(
("submit_failed:", "option_submit_failed:", "order_identity_unavailable:")
)
),
None,
)
Expand Down
40 changes: 39 additions & 1 deletion application/ibkr_order_execution.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,24 @@
from quant_platform_kit.common.models import ExecutionReport, OrderIntent
from quant_platform_kit.ibkr.execution import submit_order_intent as _submit_order_intent

from application.broker_reconciliation import (
IBKRReconciliationReadError,
build_canonical_order_key,
)

DEFAULT_TIME_IN_FORCE = "DAY"
_ORDER_IDENTITY_REQUIRED_STATUSES = frozenset(
{
"ApiPending",
"ApiPendingSubmit",
"Filled",
"Partial",
"PartiallyFilled",
"PendingSubmit",
"PreSubmitted",
"Submitted",
}
)


def _stock_factory_for_market(
Expand Down Expand Up @@ -106,7 +123,7 @@ def submit_order_intent(
"""Submit an IBKR order with explicit TIF to avoid account-preset rejections."""

intent = _intent_with_default_time_in_force(order_intent)
return _submit_order_intent(
report = _submit_order_intent(
ib,
intent,
account_id=account_id,
Expand All @@ -125,3 +142,24 @@ def submit_order_intent(
),
limit_order_factory=limit_order_factory,
)
raw_payload = dict(report.raw_payload or {})
if report.broker_order_id is None and report.status not in _ORDER_IDENTITY_REQUIRED_STATUSES:
return report
try:
order_key = build_canonical_order_key(
account_id=raw_payload.get("account_id"),
client_id=getattr(getattr(ib, "client", None), "clientId", None),
order_id=report.broker_order_id,
perm_id=raw_payload.get("perm_id"),
)
except IBKRReconciliationReadError:
return replace(
report,
status="ReconciliationRequired",
raw_payload={
**raw_payload,
"broker_status": report.status,
"reconciliation_outcome": "order_identity_unavailable",
},
)
return replace(report, raw_payload={**raw_payload, "order_key": order_key})
21 changes: 21 additions & 0 deletions docs/ibkr_reconciliation_baseline_enrollment.zh-CN.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,27 @@
让操作者人工确认“恢复原有实盘基线”。现有自动化权限策略将这类 broker/order
execution 变更视为高风险,禁止自动恢复。

## Canonical order-key 身份契约

固定依赖 `ib-insync==0.9.86` 的 `Wrapper.orderKey` 规则是:`orderId <= 0` 代表
TWS manual order,使用 `permId`;正 `orderId` 使用 `(clientId, orderId)`。平台在此
基础上加入账户隔离,形成以下唯一身份状态表:

| 输入 | canonical key | 对账结果 |
|---|---|---|
| `orderId > 0`,且 account/client/order 完整 | `ibkr|account=...|client_id=...|order_id=...` | `IDENTIFIED` |
| `orderId <= 0`,且 account/正 `permId` 完整 | `ibkr|account=...|perm_id=...` | `IDENTIFIED` |
| account、orderId、所需 clientId/permId 缺失或不可归一 | 不生成 key | `IDENTITY_UNAVAILABLE`,失败关闭 |

提交适配器只会把 `IDENTIFIED` 报告交给成功状态;身份失败固定返回
`ReconciliationRequired` 和 `reconciliation_outcome=order_identity_unavailable`,上游执行摘要
必须为 `blocked`,不得写成 `executed` 或把 key 置为 `None` 后继续。只读对账遇到同类失败时
直接拒绝本次采集,`/reconcile` 返回失败且不生成候选。

canonical key 会进入私有 open-order/recent-execution 归一记录,因此新代码生成的对应摘要可能
与旧 expected digest 不同;不匹配时继续保持 `RECONCILE_ONLY`。本契约不更新基线、不授权重录,
也不定义 partial、cancel 或 terminal outcome 语义。

## 发布给统一管理站点的最小来源快照

`scripts/publish_reconciliation_recovery_source.py` 只接受上一步的私有候选和
Expand Down
51 changes: 47 additions & 4 deletions tests/test_broker_reconciliation.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,7 @@

import pytest

import application.broker_reconciliation as broker_reconciliation
from application.broker_reconciliation import (
IBKRReconciliationObservations,
IBKRReconciliationReadError,
Expand Down Expand Up @@ -32,11 +33,13 @@ def _snapshot(*, account_id: str = "U123"):
)


def _trade(*, account_id: str = "U123"):
def _trade(*, account_id: str = "U123", order_id: int = 456, perm_id: int = 9001):
return SimpleNamespace(
order=SimpleNamespace(
account=account_id,
permId=456,
clientId=7,
orderId=order_id,
permId=perm_id,
action="BUY",
orderType="LMT",
totalQuantity=2,
Expand All @@ -54,12 +57,14 @@ def _trade(*, account_id: str = "U123"):
)


def _fill(*, account_id: str = "U123"):
def _fill(*, account_id: str = "U123", order_id: int = 456, perm_id: int = 9001):
return SimpleNamespace(
execution=SimpleNamespace(
acctNumber=account_id,
clientId=7,
execId="exec-1",
orderId=456,
orderId=order_id,
permId=perm_id,
time="20260830 13:30:00 UTC",
side="BOT",
shares=1,
Expand Down Expand Up @@ -118,6 +123,44 @@ def fetch_portfolio_snapshot(_ib, **kwargs):
assert len(observations.open_orders) == 1
assert len(observations.recent_executions) == 1
assert observations.open_orders[0]["account"] == "U123"
assert observations.open_orders[0]["order_key"] == observations.recent_executions[0]["order_key"]


@pytest.mark.parametrize("order_id", [-7, 0])
def test_manual_order_key_uses_perm_id_for_non_positive_order_ids(order_id: int) -> None:
assert broker_reconciliation.build_canonical_order_key(
account_id="U123",
client_id=7,
order_id=order_id,
perm_id=9001,
) == "ibkr|account=U123|perm_id=9001"


def test_api_order_key_uses_account_client_and_positive_order_id() -> None:
assert broker_reconciliation.build_canonical_order_key(
account_id="U123",
client_id=7,
order_id=456,
perm_id=9001,
) == "ibkr|account=U123|client_id=7|order_id=456"


@pytest.mark.parametrize(
"identity",
[
{"account_id": "U123", "client_id": 7, "order_id": 0, "perm_id": None},
{"account_id": "U123", "client_id": 7, "order_id": -7, "perm_id": 0},
{"account_id": "U123", "client_id": 7, "order_id": None, "perm_id": 9001},
{"account_id": "U123", "client_id": 7, "order_id": "invalid", "perm_id": 9001},
{"account_id": "U123", "client_id": None, "order_id": 456, "perm_id": 9001},
{"account_id": "U123", "client_id": "invalid", "order_id": 456, "perm_id": 9001},
{"account_id": "U123", "client_id": 7, "order_id": 0, "perm_id": "invalid"},
{"account_id": "", "client_id": 7, "order_id": 456, "perm_id": 9001},
],
)
def test_order_key_rejects_missing_or_unusable_identity(identity: dict[str, object]) -> None:
with pytest.raises(IBKRReconciliationReadError, match="order identity"):
broker_reconciliation.build_canonical_order_key(**identity)


def test_cash_reconciliation_ignores_dynamic_margin_and_valuation_tags() -> None:
Expand Down
Loading