富途 OpenAPI Python SDK 量化交易系统构建指南

实战案例

案例一：智能股票筛选系统

基于财务指标和价格条件，自动筛选符合要求的投资标的：

def advanced_stock_filter(api_ip, api_port):
    quote_ctx = ft.OpenQuoteContext(host=api_ip, port=api_port)
    price_filter = ft.SimpleFilter()
    price_filter.stock_field = ft.StockField.CUR_PRICE
    price_filter.filter_min = 10
    price_filter.filter_max = 200
    ret, data = quote_ctx.get_stock_filter(market=ft.Market.HK, filter_list=[price_filter])
    if ret == ft.RET_OK:
        print(f"找到 {len(data[2])} 只符合条件的股票")
    quote_ctx.close()

案例二：实时行情推送系统

构建实时监控系统，及时捕捉市场变化：

class RealTimeQuoteHandler(ft.StockQuoteHandlerBase):
    def on_recv_rsp(self, rsp_pb):
        ret_code, content = super().on_recv_rsp(rsp_pb)
        if ret_code == ft.RET_OK:
            stock_code = content['code']
            price = content['cur_price']
            print(f"实时价格更新：{stock_code} - {price}")

高级技巧

连接管理与错误处理

稳定的连接是交易系统的基础，以下技巧确保你的系统始终在线：

quote_ctx.reconnect_interval = 5

数据格式转换与处理

利用 pandas 进行高效数据处理：

import pandas as pd
kline_data = quote_ctx.get_cur_kline('HK.00700', num=100, ktype=ft.KLType.K_DAY)
if kline_data[0] == ft.RET_OK:
    df = pd.DataFrame(kline_data[1])
    print(df.head())

常见问题

连接失败排查步骤

当遇到连接问题时，按照以下顺序检查：

1. 确认 FutuOpenD 网关客户端正在运行
2. 检查 IP 地址和端口配置是否正确
3. 验证网络连接是否正常
4. 检查防火墙设置是否阻挡连接

数据获取异常处理

确保你的代码能够优雅处理各种异常情况：

try:
    result = quote_ctx.get_market_snapshot(['HK.00700'])
except Exception as e:
    print(f"数据获取异常：{e}")

进阶功能

自定义回调处理器

创建个性化的数据处理逻辑：

class CustomTickerHandler(ft.TickerHandlerBase):
    def on_recv_rsp(self, rsp_pb):
        process_tick_data(rsp_pb)
        return ft.RET_OK, "处理成功"

性能优化技巧

对于高频交易场景，这些优化将显著提升系统性能：

• 合理设置订阅数据类型，避免不必要的数据传输
• 使用异步处理机制，提高系统吞吐量
• 定期清理无用连接，释放系统资源