跳转到内容

接收家庭代码

reqFamilyCodes() 的结果通过 familyCodes() 回调返回。Python API 中回调参数是一个 FamilyCode 对象列表。

def familyCodes(self, familyCodes):
...

每个 FamilyCode 对象包含账户号和家庭代码字符串:

字段中文含义说明
accountID账户代码真实账户号,公开输出时要脱敏
familyCodeStr家庭代码字符串账户不属于 account family 时可能为空

收到回调并不代表每个账户都有非空 family code。普通账户或没有家庭结构的模拟账户,可能返回:

accountID = "DUxxxxxxx"
familyCodeStr = ""

这类结果应理解为“账户行存在,但家庭代码为空”。程序里可以这样判断:

family_code = item.familyCodeStr or "EMPTY"

不要把空字符串直接当成 API 失败。真正需要排查的是:没有收到 familyCodes() 回调,或者出现非信息类错误。

下面示例会读取 FamilyCode.accountIDFamilyCode.familyCodeStr,但输出时把账户号替换成 ACCOUNT_1 这种别名。

import threading
from ibapi.client import EClient
from ibapi.wrapper import EWrapper
INFO_CODES = {2100, 2104, 2106, 2158}
class ReceiveFamilyCodesApp(EWrapper, EClient):
def __init__(self):
EClient.__init__(self, self)
self.ready = threading.Event()
self.family_ready = threading.Event()
self.family_codes = []
self.info_codes = []
self.errors_seen = []
def nextValidId(self, orderId):
self.ready.set()
def familyCodes(self, familyCodes):
self.family_codes = list(familyCodes or [])
self.family_ready.set()
def error(self, reqId, errorTime, errorCode, errorString, advancedOrderRejectJson=""):
if errorCode in INFO_CODES:
self.info_codes.append(errorCode)
else:
self.errors_seen.append((reqId, errorCode, errorString))
app = ReceiveFamilyCodesApp()
try:
app.connect("127.0.0.1", 7497, clientId=980)
thread = threading.Thread(target=app.run, daemon=True)
thread.start()
if not app.ready.wait(8):
raise RuntimeError("等待 nextValidId 超时")
app.reqFamilyCodes()
if not app.family_ready.wait(8):
raise RuntimeError("等待 familyCodes 回调超时")
finally:
if app.isConnected():
app.disconnect()
rows = []
for index, item in enumerate(app.family_codes, 1):
rows.append({
"alias": f"ACCOUNT_{index}",
"has_account_id": bool(item.accountID),
"family_code": item.familyCodeStr or "EMPTY",
})
print(f"FAMILY_CALLBACK_RECEIVED={app.family_ready.is_set()}")
print(f"FAMILY_CODE_ROW_COUNT={len(app.family_codes)}")
print("ROWS=" + (str(rows) if rows else "EMPTY"))
print("INFO_CODES=" + (",".join(map(str, sorted(set(app.info_codes)))) if app.info_codes else "NONE"))
print(f"NON_INFO_ERROR_COUNT={len(app.errors_seen)}")

脱敏后的参考输出如下:

FAMILY_CALLBACK_RECEIVED=True
FAMILY_CODE_ROW_COUNT=1
ROWS=[{'alias': 'ACCOUNT_1', 'has_account_id': True, 'family_code': 'EMPTY'}]
INFO_CODES=2104,2106,2158
NON_INFO_ERROR_COUNT=0
IS_CONNECTED_AFTER_DISCONNECT=False

这说明账户行被返回,但家庭代码为空。开发时可以把它显示为“无家庭代码”或“不属于 account family”,不要显示成错误。

如果系统里需要保存家庭代码,可以存成这样的结构:

family_by_account = {}
for item in family_codes:
family_by_account[item.accountID] = item.familyCodeStr or None

页面展示或日志输出时再做脱敏:

display_row = {
"account": "ACCOUNT_1",
"family_code": family_by_account[real_account_id] or "无家庭代码",
}

先看是否收到回调。如果收到 familyCodes([]),说明 TWS 已响应,只是这个会话没有可返回的家庭代码行。再结合账户类型判断是否正常。

这很常见。普通账户不一定属于 account family。只要没有非信息类错误,就可以按“无家庭代码”处理。

可以用家庭代码代替账户号吗?

Section titled “可以用家庭代码代替账户号吗?”

不可以。家庭代码是账户结构辅助信息,不是下单、查账户摘要、查持仓时使用的账户代码。下单和账户接口仍应使用真实账户号。

IBKR Campus: TWS API Documentation