Fosun Wealth OpenAPI Python SDK

复星财富 OpenAPI 的 Python 客户端封装，支持会话鉴权（ECDH）、请求签名与响应解密，提供账户、交易、组合、行情等接口。

环境要求

• Python 3.8+
• 依赖：requests、cryptography

安装

从源码安装

如果希望集成到您的业务代码中，也可以将 openapi-python-sdk 目录复制到您的业务项目，并在您的业务代码目录下执行下面的安装命令：

下面是一个完整的样例目录结构，外层为业务工程目录：

mybusiness/
├── example.py   # 示例代码（业务侧调用）
├── openapi-python-sdk/
│   ├── fsopenapi/             # SDK 主代码包
│   │   ├── __init__.py
│   │   ├── client.py
│   │   ├── ...                # 其他模块
│   ├── README.md
│   ├── pyproject.toml
│   ├── setup.py
│   └── ...
└── ...

安装和调用方式举例：

在 mybusiness 目录下执行：

pip install -e ./openapi-python-sdk

在 example.py (可拷贝openapi-python-sdk下example.py)中即可导入并调用 SDK：

from fsopenapi import SDKClient
# ...

pip install -e ./openapi-python-sdk

将自动安装依赖 requests、cryptography。

注意：SDK 通过环境变量获取密钥内容。如果未检测到密钥，将直接抛出异常并停止运行。
请在启动前配置以下环境变量（例如在 .env 文件或操作系统环境中设置）：

• FSOPENAPI_SERVER_PUBLIC_KEY: 设置为您的公钥内容（PEM 格式，单行/多行均可）
• FSOPENAPI_CLIENT_PRIVATE_KEY: 设置为您的私钥内容（PEM 格式，单行/多行均可）
• SDK_TYPE: 可选；默认走 /api 前缀，设置为 ops 时走 /api/ops 前缀

例如（Linux/macOS bash）：

export FSOPENAPI_SERVER_PUBLIC_KEY="$(cat ./server_public.pem)"
export FSOPENAPI_CLIENT_PRIVATE_KEY="$(cat ./client_private.pem)"
export SDK_TYPE=ops

如果密钥未通过环境变量正确提供，SDK 初始化会抛出错误并提示密钥缺失。

如果不设置 SDK_TYPE，SDK 默认请求 /api/v1/...； 如果设置 SDK_TYPE=ops，SDK 会自动请求 /api/ops/v1/...，业务代码仍然只需要调用 SDK 提供的方法，无需手动拼接前缀。

快速开始

import logging

from fsopenapi import APIError, AuthenticationError, SDKClient

BASE_URL = "https://your-gateway-host"  # 网关 base_url，不含末尾 /
API_KEY = "your-api-key"

logging.basicConfig(level=logging.INFO)

client = SDKClient(BASE_URL, API_KEY, logging_enable=True)

# 会话由 SDK 自动管理（ECDH 握手、续期），首次调用业务接口时会自动建连
# 查询账户列表
accounts = client.account.list_accounts()
print(accounts)

# 查询持仓（可选 sub_account_id、分页等）
holdings = client.portfolio.get_holdings(sub_account_id="your-sub-account-id", start=0, count=20)
print(holdings)

# 行情（不加密）
quote = client.market.quote(codes=["hk00700", "usAAPL"])
print(quote)

日志接入

SDK 使用 Python 标准 logging，默认不会主动配置全局日志系统，也不会自动创建日志文件。

• 如果业务方已经有自己的日志体系，建议直接传入自定义 logger
• 如果业务方只想快速查看 SDK 日志，可在应用入口自行配置 logging
• SDK 默认只输出安全摘要日志，不会记录完整 api_key、密钥、签名、nonce、完整请求/响应 body
• logging_enable=True 时才会输出 SDK 请求/会话日志
• log_body=True 会在日志中追加脱敏后的请求/响应体，默认关闭

1. 使用业务方自己的 logger

import logging

from fsopenapi import JsonFormatter, SDKClient

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())

sdk_logger = logging.getLogger("myapp.fsopenapi")
sdk_logger.setLevel(logging.INFO)
sdk_logger.addHandler(handler)
sdk_logger.propagate = False

client = SDKClient(BASE_URL, API_KEY, logger=sdk_logger)

2. 使用 SDK 默认命名 logger

import logging

logging.basicConfig(level=logging.INFO)
logging.getLogger("fsopenapi").setLevel(logging.INFO)

client = SDKClient(BASE_URL, API_KEY, logging_enable=True)

3. 显式控制请求日志粒度

client = SDKClient(
    BASE_URL,
    API_KEY,
    logger=sdk_logger,
    logging_enable=True,
    log_body=True,
)

• logging_enable=False：即使业务方配置了 logger，SDK 也不会主动输出请求/会话日志
• logging_enable=True, log_body=False：输出安全摘要日志，适合生产默认配置
• logging_enable=True, log_body=True：输出脱敏后的请求/响应体，适合排查问题时临时开启

4. log_body=True 示例

import logging

from fsopenapi import JsonFormatter, SDKClient

handler = logging.StreamHandler()
handler.setFormatter(JsonFormatter())

sdk_logger = logging.getLogger("myapp.fsopenapi.debug")
sdk_logger.setLevel(logging.INFO)
sdk_logger.addHandler(handler)
sdk_logger.propagate = False

client = SDKClient(
    BASE_URL,
    API_KEY,
    logger=sdk_logger,
    logging_enable=True,
    log_body=True,
)

client.trade.create_order(
    sub_account_id="sub-xxx",
    stock_code="00700",
    direction=1,
    order_type=1,
    quantity="100",
    price="350.00",
    market_code="hk",
    currency="HKD",
)

开启后，日志中会额外带上脱敏后的 request_body、query_params、response_body 或 response_text。建议仅在排查问题时临时开启，排查完成后关闭。

常见日志事件包括：

• session_create_start
• session_create_success
• session_restore_success
• session_refresh_required
• request_start
• request_success
• response_business_error
• response_decrypt_failed
• request_http_error

模块说明

模块	说明
client.session	会话管理：创建/查询/删除会话、本地会话状态
client.account	账户：交易账户列表
client.portfolio	组合：资金汇总、持仓
client.trade	交易：下单、改单、撤单、订单列表、资金流水、买卖信息、订阅管理
client.market	行情：报价、K 线、分时、盘口、逐笔、经纪队列

使用示例

1. 会话管理

# 创建会话（通常由 SDK 内部自动调用）
session_info = client.session.create_session()
print(session_info)  # sessionId, serverPublicKey, expiresIn, expiresAt

# 查询服务端当前会话
current = client.session.query_session()
print(current)

# 检查本地会话是否有效
if client.session.is_session_valid():
    print("会话有效")

# 主动登出
client.session.delete_session()

2. 账户

# 账户列表（空 body）
accounts = client.account.list_accounts()

3. 组合

# 资金汇总（可按 sub_account_id、client_id、currency 筛选）
summary = client.portfolio.get_assets_summary(sub_account_id="sub-xxx")

# 持仓列表（分页、产品类型、币种、标的等）
holdings = client.portfolio.get_holdings(
    sub_account_id="sub-xxx",
    start=0,
    count=100,
    product_types=[1, 2],
    currencies=["HKD", "USD"],
)

4. 交易

# 下单（限价单示例）
order = client.trade.create_order(
    sub_account_id="sub-xxx",
    stock_code="00700",
    direction=1,       # 1 买 2 卖
    order_type=1,     # 限价
    quantity="100",
    price="350.00",
    market_code="hk",
    currency="HKD",
)
print(order)

# 撤单
client.trade.cancel_order(order_id="order-xxx", sub_account_id="sub-xxx")

# 改单
modified = client.trade.order_modify(
    sub_account_id="sub-xxx",
    order_id="order-xxx",
    modify_type=1,  # 1 修改普通订单参数，2 修改条件单参数
    quantity="100",
    price="351.00",
)
print(modified)

# 订单列表（分页、状态、日期、方向等;show_type 0=正股 1=正股+期权 2=只期权）
orders = client.trade.list_orders(
    sub_account_id="sub-xxx",
    start=0,
    count=20,
    status_arr=[20, 40],
    from_date="2025-01-01",
    to_date="2025-01-31",
    show_type=2,  # 仅查期权时传 2
)

# 资金流水
flows = client.trade.get_cash_flows(
    sub_account_id="sub-xxx",
    trade_date_from="2025-01-01",
    trade_date_to="2025-01-31",
)

# 买卖信息（下单前校验）；期权时传 product_type=15、expiry、strike、right
bid_ask = client.trade.get_bid_ask_info(
    sub_account_id="sub-xxx",
    stock_code="00700",
    order_type=3,
    market_code="hk",
    quantity="100",
    direction=1,
    product_type=15, expiry="20260327", strike="125.00", right="CALL",  # 期权
)

# 创建交易订阅（当前仅暴露 `orderUpdate`，`channel_type=1` 表示 HTTP Webhook）
subscription = client.trade.create_subscription(
    event_type="orderUpdate",
    endpoint="https://your-partner-host/webhook",
)
print(subscription)

# 查询订阅列表
subscriptions = client.trade.list_subscriptions(start=0, count=20)
print(subscriptions)

# 按事件类型过滤（当前仅支持 orderUpdate）
filtered = client.trade.list_subscriptions(
    start=0,
    count=20,
    event_type="orderUpdate",
)
print(filtered)

# 更新订阅回调地址
updated = client.trade.update_subscription(
    subscription_id=123456,
    endpoint="https://your-partner-host/webhook/v2",
)
print(updated)

# 删除订阅
deleted = client.trade.delete_subscription(subscription_id=123456)
print(deleted)

5. 行情

# 批量报价（codes 为 marketCode+stockCode，如 hk00700、usAAPL）
quote = client.market.quote(codes=["hk00700", "usAAPL"])

# K 线（code 格式: marketCode + stockCode，如 hk00700）
klines = client.market.kline("hk00700", ktype="day", num=30)

# 分时
min_data = client.market.min("hk00700", count=5)

# 盘口/买卖档
orderbook = client.market.orderbook("hk00700", count=5)

# 逐笔成交
ticks = client.market.tick("hk00700", count=20, id=-1)

# 买卖盘经纪商队列
brokers = client.market.broker_list("hk00700")

异常处理

from fsopenapi import APIError, AuthenticationError, PermissionError, CacheError

try:
    data = client.account.list_accounts()
except AuthenticationError as e:
    print(f"鉴权失败: {e.code} - {e.message}")
except PermissionError as e:
    print(f"权限错误: {e.code} - {e.message}")
except APIError as e:
    print(f"接口错误: {e.code} - {e.message} (requestId: {e.request_id})")

注意事项

1. base_url：填写网关完整 base_url（如 https://host），不要以 / 结尾。
2. API Key：由开放平台下发，请勿泄露。
3. 会话：SDK 会自动完成 ECDH 建连与过期前续期，业务侧一般无需手动调用 session.create_session()。
4. 行情接口：走 /market/，仅签名不加密；交易/账户等接口请求体与响应支持 AES-GCM 加解密。