LongPort OpenAPI Python SDK 开发指南

本文档为 AI IDE 上下文优化，提供 LongPort OpenAPI Python SDK 的完整接口定义、使用方法和最佳实践。

快速开始

安装

pip install longport

# 使用阿里云镜像（可选）
pip install longport -i https://mirrors.aliyun.com/pypi/simple

基本使用流程

配置环境变量
创建 Config 对象
创建 Context（QuoteContext 或 TradeContext）
调用 API 方法

认证和配置

Config 类

SDK 的核心配置类，管理所有认证信息和连接参数。

类定义

class Config:
    def __init__(
        self,
        app_key: str,                                    # App Key（必需）
        app_secret: str,                                 # App Secret（必需）
        access_token: str,                               # Access Token（必需）
        http_url: Optional[str] = None,                 # HTTP API URL
        quote_ws_url: Optional[str] = None,             # 行情 WebSocket URL
        trade_ws_url: Optional[str] = None,             # 交易 WebSocket URL
        language: Optional[Type[Language]] = None,       # 语言（Language.ZH_CN/ZH_HK/EN）
        enable_overnight: bool = False,                  # 启用夜盘行情
        push_candlestick_mode: Type[PushCandlestickMode] = PushCandlestickMode.Realtime,
        enable_print_quote_packages: bool = True,        # 打印行情包信息
        log_path: Optional[str] = None                   # 日志文件路径
    ) -> None

创建配置

方法一：从环境变量创建（推荐）

from longport.openapi import Config

# 从环境变量自动加载
config = Config.from_env()

环境变量列表：

# 必需变量
export LONGPORT_APP_KEY="your_app_key"
export LONGPORT_APP_SECRET="your_app_secret"
export LONGPORT_ACCESS_TOKEN="your_access_token"

# 可选变量
export LONGPORT_LANGUAGE="zh-CN"              # zh-CN, zh-HK, en
export LONGPORT_HTTP_URL="https://..."        # 自定义 HTTP URL
export LONGPORT_QUOTE_WS_URL="wss://..."      # 自定义行情 WebSocket URL
export LONGPORT_TRADE_WS_URL="wss://..."      # 自定义交易 WebSocket URL
export LONGPORT_ENABLE_OVERNIGHT="true"       # 启用夜盘（true/false）
export LONGPORT_PUSH_CANDLESTICK_MODE="realtime"  # realtime/confirmed
export LONGPORT_PRINT_QUOTE_PACKAGES="true"   # 打印行情包（true/false）
export LONGPORT_LOG_PATH="/path/to/log"       # 日志路径

方法二：代码直接创建

from longport.openapi import Config, Language

config = Config(
    app_key="your_app_key",
    app_secret="your_app_secret",
    access_token="your_access_token",
    language=Language.ZH_CN,
    enable_overnight=True  # 启用夜盘行情
)

Access Token 刷新

# 刷新 access_token（默认 90 天有效期）
from datetime import datetime, timedelta

new_token = config.refresh_access_token(
    expired_at=datetime.now() + timedelta(days=90)
)
print(f"New token: {new_token}")

HttpClient 类

用于直接调用 HTTP API 的客户端。

from longport.openapi import HttpClient

# 从环境变量创建
http_client = HttpClient.from_env()

# 或手动创建
http_client = HttpClient(
    http_url="https://openapi.longportapp.com",
    app_key="your_app_key",
    app_secret="your_app_secret",
    access_token="your_access_token"
)

# 发送请求
resp = http_client.request("get", "/v1/trade/execution/today")
print(resp)

# POST 请求
resp = http_client.request(
    "post",
    "/v1/trade/order/submit",
    body={"symbol": "700.HK", "order_type": "LO"}
)

核心接口

QuoteContext - 行情上下文

负责所有行情相关操作，包括订阅、查询、K线数据等。

初始化

from longport.openapi import Config, QuoteContext

config = Config.from_env()
ctx = QuoteContext(config)

# 获取会员信息
member_id = ctx.member_id()
quote_level = ctx.quote_level()
print(f"Member ID: {member_id}, Quote Level: {quote_level}")

TradeContext - 交易上下文

负责所有交易相关操作，包括下单、查询订单、账户信息等。

初始化

from longport.openapi import Config, TradeContext

config = Config.from_env()
ctx = TradeContext(config)

行情 API

订阅机制

LongPort SDK 使用推送订阅模式，需要先设置回调函数，然后订阅标的。

订阅行情（实时报价）

from time import sleep
from longport.openapi import Config, QuoteContext, SubType, PushQuote

def on_quote(symbol: str, event: PushQuote):
    """行情回调函数"""
    print(f"Symbol: {symbol}")
    print(f"  最新价: {event.last_done}")
    print(f"  开盘价: {event.open}")
    print(f"  最高价: {event.high}")
    print(f"  最低价: {event.low}")
    print(f"  成交量: {event.volume}")
    print(f"  成交额: {event.turnover}")
    print(f"  交易状态: {event.trade_status}")
    print(f"  交易时段: {event.trade_session}")

config = Config.from_env()
ctx = QuoteContext(config)

# 设置回调
ctx.set_on_quote(on_quote)

# 订阅（is_first_push=True 立即推送当前数据）
ctx.subscribe(
    symbols=["700.HK", "AAPL.US", "TSLA.US"],
    sub_types=[SubType.Quote],
    is_first_push=True
)

# 持续接收推送
sleep(30)

# 取消订阅
ctx.unsubscribe(["AAPL.US"], [SubType.Quote])

订阅深度（买卖盘）

from longport.openapi import SubType, PushDepth

def on_depth(symbol: str, event: PushDepth):
    """深度回调函数"""
    print(f"{symbol} 深度数据:")
    print("卖盘 (Asks):")
    for depth in event.asks:
        print(f"  价格: {depth.price}, 数量: {depth.volume}")
    print("买盘 (Bids):")
    for depth in event.bids:
        print(f"  价格: {depth.price}, 数量: {depth.volume}")

ctx.set_on_depth(on_depth)
ctx.subscribe(["700.HK"], [SubType.Depth], is_first_push=True)

订阅经纪商队列

from longport.openapi import SubType, PushBrokers

def on_brokers(symbol: str, event: PushBrokers):
    """经纪商队列回调"""
    print(f"{symbol} 经纪商队列:")
    print("卖方经纪商:")
    for broker in event.ask_brokers:
        print(f"  档位: {broker.position}, 经纪商ID: {broker.broker_ids}")
    print("买方经纪商:")
    for broker in event.bid_brokers:
        print(f"  档位: {broker.position}, 经纪商ID: {broker.broker_ids}")

ctx.set_on_brokers(on_brokers)
ctx.subscribe(["700.HK"], [SubType.Brokers], is_first_push=True)

订阅逐笔交易

from longport.openapi import SubType, PushTrades

def on_trades(symbol: str, event: PushTrades):
    """逐笔交易回调"""
    print(f"{symbol} 交易记录:")
    for trade in event.trades:
        print(f"  时间: {trade.timestamp}")
        print(f"  价格: {trade.price}")
        print(f"  数量: {trade.volume}")
        print(f"  方向: {trade.trade_type}")

ctx.set_on_trades(on_trades)
ctx.subscribe(["700.HK"], [SubType.Trade], is_first_push=False)

订阅 K 线

from longport.openapi import Period, PushCandlestick, TradeSessions

def on_candlestick(symbol: str, event: PushCandlestick):
    """K线更新回调"""
    candle = event.candlestick
    print(f"{symbol} K线更新:")
    print(f"  周期: {event.period}")
    print(f"  时间: {candle.timestamp}")
    print(f"  开: {candle.open}, 高: {candle.high}")
    print(f"  低: {candle.low}, 收: {candle.close}")
    print(f"  成交量: {candle.volume}")
    print(f"  是否确认: {event.is_confirmed}")

ctx.set_on_candlestick(on_candlestick)

# 订阅 1 分钟 K 线
candlesticks = ctx.subscribe_candlesticks(
    symbol="700.HK",
    period=Period.Min_1,
    trade_sessions=TradeSessions.Intraday
)
print(f"返回历史 K 线数量: {len(candlesticks)}")

# 取消订阅
ctx.unsubscribe_candlesticks("700.HK", Period.Min_1)

查询订阅信息

# 获取当前订阅列表
subscriptions = ctx.subscriptions()
for sub in subscriptions:
    print(f"Symbol: {sub.symbol}")
    print(f"  订阅类型: {sub.sub_types}")
    print(f"  K线周期: {sub.candlesticks}")

获取证券信息

基本信息（静态数据）

from longport.openapi import QuoteContext, Config

config = Config.from_env()
ctx = QuoteContext(config)

# 获取证券基本信息
securities = ctx.static_info(["700.HK", "AAPL.US"])
for sec in securities:
    print(f"代码: {sec.symbol}")
    print(f"  中文名: {sec.name_cn}")
    print(f"  英文名: {sec.name_en}")
    print(f"  交易所: {sec.exchange}")
    print(f"  货币: {sec.currency}")
    print(f"  每手股数: {sec.lot_size}")
    print(f"  总股本: {sec.total_shares}")
    print(f"  流通股: {sec.circulating_shares}")
    print(f"  EPS: {sec.eps}")
    print(f"  每股净资产: {sec.bps}")
    print(f"  股息率: {sec.dividend_yield}")
    print(f"  板块: {sec.board}")

实时行情

# 获取实时行情（不需要订阅）
quotes = ctx.quote(["700.HK", "AAPL.US", "TSLA.US"])
for q in quotes:
    print(f"{q.symbol}:")
    print(f"  最新价: {q.last_done}")
    print(f"  昨收: {q.prev_close}")
    print(f"  开盘: {q.open}")
    print(f"  最高: {q.high}")
    print(f"  最低: {q.low}")
    print(f"  成交量: {q.volume}")
    print(f"  成交额: {q.turnover}")
    print(f"  交易状态: {q.trade_status}")

    # 美股盘前盘后数据
    if q.pre_market_quote:
        print(f"  盘前价: {q.pre_market_quote.last_done}")
    if q.post_market_quote:
        print(f"  盘后价: {q.post_market_quote.last_done}")
    if q.overnight_quote:
        print(f"  夜盘价: {q.overnight_quote.last_done}")

期权行情

# 获取期权行情
option_quotes = ctx.option_quote(["AAPL230317P160000.US"])
for opt in option_quotes:
    print(f"{opt.symbol}:")
    print(f"  最新价: {opt.last_done}")
    print(f"  隐含波动率: {opt.implied_volatility}")
    print(f"  未平仓合约: {opt.open_interest}")
    print(f"  到期日: {opt.expiry_date}")
    print(f"  行权价: {opt.strike_price}")
    print(f"  合约乘数: {opt.contract_multiplier}")
    print(f"  期权类型: {opt.contract_type}")  # American/Europe
    print(f"  方向: {opt.direction}")  # Call/Put

权证行情

# 获取权证行情（港股）
warrant_quotes = ctx.warrant_quote(["21125.HK"])
for w in warrant_quotes:
    print(f"{w.symbol}:")
    print(f"  最新价: {w.last_done}")
    print(f"  隐含波动率: {w.implied_volatility}")
    print(f"  到期日: {w.expiry_date}")
    print(f"  最后交易日: {w.last_trade_date}")
    print(f"  街货比: {w.outstanding_ratio}")

深度和经纪商数据

# 获取深度数据
depth = ctx.depth("700.HK")
print(f"卖盘深度:")
for ask in depth.asks:
    print(f"  {ask.position}: 价格 {ask.price}, 数量 {ask.volume}")
print(f"买盘深度:")
for bid in depth.bids:
    print(f"  {bid.position}: 价格 {bid.price}, 数量 {bid.volume}")

# 获取经纪商队列
brokers = ctx.brokers("700.HK")
print(f"卖方经纪商:")
for broker in brokers.ask_brokers:
    print(f"  档位 {broker.position}: {broker.broker_ids}")
print(f"买方经纪商:")
for broker in brokers.bid_brokers:
    print(f"  档位 {broker.position}: {broker.broker_ids}")

逐笔交易

# 获取最近 N 笔交易（最大 1000）
trades = ctx.trades("700.HK", count=10)
for trade in trades:
    print(f"时间: {trade.timestamp}")
    print(f"  价格: {trade.price}")
    print(f"  数量: {trade.volume}")
    print(f"  成交额: {trade.turnover}")
    print(f"  方向: {trade.trade_type}")  # Neutral/Up/Down
    print(f"  交易时段: {trade.trade_session}")

K 线数据

获取最近 K 线

from longport.openapi import Period, AdjustType, TradeSessions

# 获取最近 N 条 K 线（最大 1000）
candlesticks = ctx.candlesticks(
    symbol="700.HK",
    period=Period.Day,          # 日线
    count=10,
    adjust_type=AdjustType.NoAdjust,  # 不复权
    trade_sessions=TradeSessions.Intraday  # 日内交易
)

for candle in candlesticks:
    print(f"{candle.timestamp}:")
    print(f"  开: {candle.open}, 高: {candle.high}")
    print(f"  低: {candle.low}, 收: {candle.close}")
    print(f"  成交量: {candle.volume}")
    print(f"  成交额: {candle.turnover}")

历史 K 线（按偏移量）

import datetime

# 从指定时间点获取 K 线
candlesticks = ctx.history_candlesticks_by_offset(
    symbol="700.HK",
    period=Period.Day,
    adjust_type=AdjustType.ForwardAdjust,  # 前复权
    forward=False,          # False: 获取指定时间之前的数据
    count=10,
    time=datetime.datetime(2023, 8, 18),
    trade_sessions=TradeSessions.Intraday
)

历史 K 线（按日期范围）

import datetime

# 获取日期范围内的 K 线
candlesticks = ctx.history_candlesticks_by_date(
    symbol="700.HK",
    period=Period.Day,
    adjust_type=AdjustType.NoAdjust,
    start=datetime.date(2022, 5, 5),
    end=datetime.date(2022, 6, 23),
    trade_sessions=TradeSessions.Intraday
)

日内分时数据

# 获取日内分时线
intraday = ctx.intraday("700.HK", TradeSessions.Intraday)
for line in intraday:
    print(f"{line.timestamp}: 价格 {line.price}, 成交量 {line.volume}")

期权链

import datetime

# 获取期权到期日列表
expiry_dates = ctx.option_chain_expiry_date_list("AAPL.US")
print(f"期权到期日: {expiry_dates}")

# 获取指定到期日的期权链
option_chain = ctx.option_chain_info_by_date(
    symbol="AAPL.US",
    expiry_date=datetime.date(2023, 1, 20)
)
for strike in option_chain:
    print(f"行权价: {strike.strike_price}")
    print(f"  Call: {strike.call_symbol}")
    print(f"  Put: {strike.put_symbol}")

权证筛选

from longport.openapi import WarrantSortBy, SortOrderType, WarrantType

# 获取权证发行商
issuers = ctx.warrant_issuers()
for issuer in issuers:
    print(f"ID: {issuer.id}, 名称: {issuer.name_cn}")

# 筛选权证列表
warrants = ctx.warrant_list(
    symbol="700.HK",
    sort_by=WarrantSortBy.LastDone,
    sort_order=SortOrderType.Ascending,
    warrant_type=[WarrantType.Call],  # 只看认购证
    issuer=[1, 2]  # 指定发行商
)
for w in warrants:
    print(f"{w.symbol}: {w.name}")

市场信息

交易日历

from datetime import date
from longport.openapi import Market

# 获取交易日
trading_days = ctx.trading_days(
    market=Market.HK,
    begin=date(2022, 1, 1),
    end=date(2022, 2, 1)
)
print(f"交易日: {trading_days.trading_days}")
print(f"半日市: {trading_days.half_trading_days}")

当日交易时段

# 获取当日各市场交易时段
sessions = ctx.trading_session()
for session in sessions:
    print(f"{session.market}:")
    for ts in session.trade_sessions:
        print(f"  {ts.beg_time} - {ts.end_time}: {ts.trade_session}")

市场温度

# 获取当前市场温度
temp = ctx.market_temperature(Market.HK)
print(f"温度值: {temp.temperature}")
print(f"描述: {temp.description}")
print(f"估值: {temp.valuation}")
print(f"情绪: {temp.sentiment}")

# 获取历史市场温度
history_temp = ctx.history_market_temperature(
    market=Market.HK,
    start=date(2023, 1, 1),
    end=date(2023, 1, 31)
)
print(f"粒度: {history_temp.granularity}")
for record in history_temp.records:
    print(f"{record.timestamp}: {record.temperature}")

资金流向

# 获取资金流向日内数据
capital_flow = ctx.capital_flow("700.HK")
for flow in capital_flow:
    print(f"{flow.timestamp}:")
    print(f"  流入: {flow.inflow}")
    print(f"  流出: {flow.outflow}")

# 获取资金分布
distribution = ctx.capital_distribution("700.HK")
print(f"时间: {distribution.timestamp}")
print(f"流入资金:")
print(f"  大单: {distribution.capital_in.large}")
print(f"  中单: {distribution.capital_in.medium}")
print(f"  小单: {distribution.capital_in.small}")

自选股

# 获取自选股列表
watchlist = ctx.watchlist()
for group in watchlist:
    print(f"分组 {group.id}: {group.name}")
    for sec in group.securities:
        print(f"  {sec.symbol}: {sec.name}")
        print(f"    关注价: {sec.watched_price}")
        print(f"    关注时间: {sec.watched_at}")

# 创建自选股分组
group_id = ctx.create_watchlist_group(
    name="我的自选",
    securities=["700.HK", "AAPL.US"]
)
print(f"创建分组 ID: {group_id}")

# 更新自选股分组
from longport.openapi import SecuritiesUpdateMode

ctx.update_watchlist_group(
    id=group_id,
    name="新名称",
    securities=["TSLA.US"],
    mode=SecuritiesUpdateMode.Add  # Add/Remove/Replace
)

# 删除自选股分组
ctx.delete_watchlist_group(id=group_id, purge=False)

计算指标

from longport.openapi import CalcIndex

# 获取计算指标
indexes = ctx.calc_indexes(
    symbols=["700.HK", "AAPL.US"],
    indexes=[
        CalcIndex.LastDone,
        CalcIndex.ChangeRate,
        CalcIndex.Volume,
        CalcIndex.Turnover,
        CalcIndex.PeTtmRatio,
        CalcIndex.PbRatio
    ]
)

for idx in indexes:
    print(f"{idx.symbol}:")
    print(f"  最新价: {idx.last_done}")
    print(f"  涨跌幅: {idx.change_rate}")
    print(f"  成交量: {idx.volume}")
    print(f"  成交额: {idx.turnover}")
    print(f"  市盈率: {idx.pe_ttm_ratio}")
    print(f"  市净率: {idx.pb_ratio}")

实时数据（从本地缓存）

订阅后，可以从本地缓存快速获取实时数据，无需等待推送。

from time import sleep

# 先订阅
ctx.subscribe(["700.HK", "AAPL.US"], [SubType.Quote], is_first_push=True)
sleep(2)

# 从本地缓存获取实时行情
realtime_quotes = ctx.realtime_quote(["700.HK", "AAPL.US"])
for q in realtime_quotes:
    print(f"{q.symbol}: {q.last_done}")

# 获取实时深度
realtime_depth = ctx.realtime_depth("700.HK")

# 获取实时经纪商
realtime_brokers = ctx.realtime_brokers("700.HK")

# 获取实时逐笔
realtime_trades = ctx.realtime_trades("700.HK", count=10)

# 获取实时 K 线
ctx.subscribe_candlesticks("AAPL.US", Period.Min_1)
sleep(2)
realtime_candles = ctx.realtime_candlesticks("AAPL.US", Period.Min_1, count=10)

交易 API

订单提交

submit_order 方法定义

def submit_order(
    self,
    side: Type[OrderSide],                      # 买卖方向（必需）
    symbol: str,                                 # 证券代码（必需）
    order_type: Type[OrderType],                # 订单类型（必需）
    submitted_quantity: Decimal,                 # 订单数量（必需）
    time_in_force: Type[TimeInForceType],       # 有效期类型（必需）
    submitted_price: Optional[Decimal] = None,   # 限价单价格
    trigger_price: Optional[Decimal] = None,     # 触发价格
    limit_offset: Optional[Decimal] = None,      # 限价偏移
    trailing_amount: Optional[Decimal] = None,   # 跟踪金额
    trailing_percent: Optional[Decimal] = None,  # 跟踪百分比
    expire_date: Optional[date] = None,          # 到期日期
    outside_rth: Optional[Type[OutsideRTH]] = None,  # 盘后交易
    remark: Optional[str] = None                 # 备注
) -> SubmitOrderResponse

基本订单类型

1. 限价单（LO - Limit Order）

from decimal import Decimal
from longport.openapi import TradeContext, Config, OrderSide, OrderType, TimeInForceType

config = Config.from_env()
ctx = TradeContext(config)

# 买入限价单
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.LO,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    time_in_force=TimeInForceType.Day,
    remark="限价买入"
)
print(f"订单ID: {resp.order_id}")

2. 增强限价单（ELO - Enhanced Limit Order）

# 增强限价单（港股特有，更好的成交概率）
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.ELO,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    time_in_force=TimeInForceType.Day
)

3. 市价单（MO - Market Order）

from longport.openapi import OutsideRTH

# 市价单（立即按市场价格成交）
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.MO,
    submitted_quantity=Decimal("200"),
    time_in_force=TimeInForceType.Day,
    outside_rth=OutsideRTH.AnyTime  # 美股可盘后交易
)

4. 竞价单（AO/ALO - At-auction Order）

# 竞价单（在集合竞价时段成交）
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.AO,
    submitted_quantity=Decimal("200"),
    time_in_force=TimeInForceType.Day
)

# 竞价限价单
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.ALO,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    time_in_force=TimeInForceType.Day
)

5. 零股单（ODD - Odd Lots）

# 零股单（交易少于一手的股票）
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.ODD,
    submitted_quantity=Decimal("50"),  # 少于一手
    submitted_price=Decimal("50.00"),
    time_in_force=TimeInForceType.Day
)

条件订单

6. 限价触发单（LIT - Limit If Touched）

# 价格达到触发价后，转为限价单
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.LIT,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),     # 限价
    trigger_price=Decimal("49.50"),       # 触发价
    time_in_force=TimeInForceType.Day
)

7. 市价触发单（MIT - Market If Touched）

# 价格达到触发价后，转为市价单
resp = ctx.submit_order(
    side=OrderSide.Sell,
    symbol="700.HK",
    order_type=OrderType.MIT,
    submitted_quantity=Decimal("200"),
    trigger_price=Decimal("48.00"),  # 触发价
    time_in_force=TimeInForceType.Day
)

跟踪订单

8. 跟踪限价触发单（TSLPAMT - Trailing Stop Limit by Amount）

# 按固定金额跟踪价格
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.TSLPAMT,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    trigger_price=Decimal("49.50"),
    trailing_amount=Decimal("0.50"),    # 跟踪金额
    limit_offset=Decimal("0.10"),       # 限价偏移
    time_in_force=TimeInForceType.Day
)

9. 跟踪限价触发单（TSLPPCT - Trailing Stop Limit by Percent）

# 按百分比跟踪价格
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.TSLPPCT,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    trigger_price=Decimal("49.50"),
    trailing_percent=Decimal("1.0"),   # 跟踪百分比（1%）
    limit_offset=Decimal("0.10"),      # 限价偏移
    time_in_force=TimeInForceType.Day
)

10. 跟踪市价触发单（TSMAMT/TSMPCT）

# 按金额跟踪的市价单
resp = ctx.submit_order(
    side=OrderSide.Sell,
    symbol="700.HK",
    order_type=OrderType.TSMAMT,
    submitted_quantity=Decimal("200"),
    trigger_price=Decimal("51.00"),
    trailing_amount=Decimal("0.50"),
    time_in_force=TimeInForceType.Day
)

# 按百分比跟踪的市价单
resp = ctx.submit_order(
    side=OrderSide.Sell,
    symbol="700.HK",
    order_type=OrderType.TSMPCT,
    submitted_quantity=Decimal("200"),
    trigger_price=Decimal("51.00"),
    trailing_percent=Decimal("1.0"),
    time_in_force=TimeInForceType.Day
)

时间有效期

# 当日有效（日内交易）
time_in_force=TimeInForceType.Day

# 撤销前有效（长期有效）
time_in_force=TimeInForceType.GoodTilCanceled

# 指定日期前有效
from datetime import date
resp = ctx.submit_order(
    side=OrderSide.Buy,
    symbol="700.HK",
    order_type=OrderType.LO,
    submitted_quantity=Decimal("200"),
    submitted_price=Decimal("50.00"),
    time_in_force=TimeInForceType.GoodTilDate,
    expire_date=date(2023, 12, 31)  # 指定到期日
)

盘后交易（美股）

# 仅常规交易时段
outside_rth=OutsideRTH.RTHOnly

# 任意时间（含盘前盘后）
outside_rth=OutsideRTH.AnyTime

# 夜盘
outside_rth=OutsideRTH.Overnight

订单管理

修改订单

# 修改订单价格和数量
ctx.replace_order(
    order_id="123456",
    quantity=Decimal("300"),           # 新数量
    price=Decimal("51.00"),            # 新价格
    trigger_price=Decimal("50.50"),    # 新触发价（可选）
    limit_offset=Decimal("0.20"),      # 新限价偏移（可选）
    trailing_amount=Decimal("0.60"),   # 新跟踪金额（可选）
    trailing_percent=Decimal("1.5"),   # 新跟踪百分比（可选）
    remark="修改订单"
)

取消订单

# 取消订单
ctx.cancel_order(order_id="123456")

订单查询

查询今日订单

# 查询今日所有订单
orders = ctx.today_orders()
for order in orders:
    print(f"订单ID: {order.order_id}")
    print(f"  代码: {order.symbol}")
    print(f"  方向: {order.side}")  # Buy/Sell
    print(f"  类型: {order.order_type}")
    print(f"  状态: {order.status}")
    print(f"  提交数量: {order.quantity}")
    print(f"  已成交: {order.executed_quantity}")
    print(f"  提交价格: {order.price}")
    print(f"  成交价格: {order.executed_price}")
    print(f"  提交时间: {order.submitted_at}")

# 按条件过滤
orders = ctx.today_orders(
    symbol="700.HK",      # 指定标的
    status=[OrderStatus.Filled, OrderStatus.PartialFilled],  # 指定状态
    side=OrderSide.Buy,   # 指定方向
    market=Market.HK,     # 指定市场
    order_id="123456"     # 指定订单ID
)

查询历史订单

from datetime import date

# 查询历史订单
orders = ctx.history_orders(
    symbol="700.HK",
    status=[OrderStatus.Filled],
    side=OrderSide.Buy,
    market=Market.HK,
    start_at=date(2023, 1, 1),
    end_at=date(2023, 12, 31)
)

查询订单详情

# 获取订单详情（含费用明细）
detail = ctx.order_detail(order_id="123456")
print(f"订单ID: {detail.order_id}")
print(f"状态: {detail.status}")
print(f"股票名称: {detail.stock_name}")
print(f"已成交: {detail.executed_quantity}")
print(f"成交价: {detail.executed_price}")
print(f"货币: {detail.currency}")

# 佣金和费用
print(f"免佣状态: {detail.free_status}")
print(f"免佣金额: {detail.free_amount}")

# 费用明细
charge = detail.charge_detail
print(f"总费用: {charge.total_amount} {charge.currency}")
for item in charge.items:
    print(f"  {item.name}:")
    for fee in item.fees:
        print(f"    {fee.name}: {fee.amount} {fee.currency}")

# 订单历史
for history in detail.history:
    print(f"{history.time}: {history.status}")
    print(f"  价格: {history.price}, 数量: {history.quantity}")
    print(f"  消息: {history.msg}")

成交查询

查询今日成交

# 查询今日所有成交
executions = ctx.today_executions()
for exe in executions:
    print(f"订单ID: {exe.order_id}")
    print(f"  成交ID: {exe.trade_id}")
    print(f"  代码: {exe.symbol}")
    print(f"  成交时间: {exe.trade_done_at}")
    print(f"  成交数量: {exe.quantity}")
    print(f"  成交价格: {exe.price}")

# 按条件过滤
executions = ctx.today_executions(
    symbol="700.HK",
    order_id="123456"
)

查询历史成交

from datetime import date

executions = ctx.history_executions(
    symbol="700.HK",
    start_at=date(2023, 1, 1),
    end_at=date(2023, 12, 31)
)

订单推送通知

from longport.openapi import TopicType, PushOrderChanged

def on_order_changed(event: PushOrderChanged):
    """订单变更回调"""
    print(f"订单变更通知:")
    print(f"  订单ID: {event.order_id}")
    print(f"  代码: {event.symbol}")
    print(f"  股票名称: {event.stock_name}")
    print(f"  方向: {event.side}")
    print(f"  订单类型: {event.order_type}")
    print(f"  状态: {event.status}")
    print(f"  提交数量: {event.submitted_quantity}")
    print(f"  已成交: {event.executed_quantity}")
    print(f"  提交价格: {event.submitted_price}")
    print(f"  成交价格: {event.executed_price}")
    print(f"  更新时间: {event.updated_at}")
    print(f"  备注: {event.msg}")

# 设置回调并订阅
ctx.set_on_order_changed(on_order_changed)
ctx.subscribe([TopicType.Private])

# 现在所有订单变更都会触发回调

账户信息

查询账户余额

# 查询所有账户余额
balances = ctx.account_balance()
for balance in balances:
    print(f"货币: {balance.currency}")
    print(f"  总现金: {balance.total_cash}")
    print(f"  净资产: {balance.net_assets}")
    print(f"  最大融资额: {balance.max_finance_amount}")
    print(f"  剩余融资额: {balance.remaining_finance_amount}")
    print(f"  风控等级: {balance.risk_level}")
    print(f"  保证金催缴: {balance.margin_call}")
    print(f"  初始保证金: {balance.init_margin}")
    print(f"  维持保证金: {balance.maintenance_margin}")
    print(f"  购买力: {balance.buy_power}")

    # 现金明细
    for cash in balance.cash_infos:
        print(f"  {cash.currency}:")
        print(f"    可提现金: {cash.withdraw_cash}")
        print(f"    可用现金: {cash.available_cash}")
        print(f"    冻结现金: {cash.frozen_cash}")
        print(f"    待交收: {cash.settling_cash}")

# 查询指定货币
balances = ctx.account_balance(currency="HKD")

查询资金流水

from datetime import date

# 查询资金流水
cash_flows = ctx.cash_flow(
    start_at=date(2023, 1, 1),
    end_at=date(2023, 12, 31),
    business_type=BalanceType.Stock,  # Stock/Cash/Fund
    symbol="700.HK",
    page=1,
    size=50
)

for flow in cash_flows:
    print(f"{flow.business_time}:")
    print(f"  名称: {flow.transaction_flow_name}")
    print(f"  方向: {flow.direction}")  # In/Out
    print(f"  业务类型: {flow.business_type}")
    print(f"  金额: {flow.balance} {flow.currency}")
    print(f"  代码: {flow.symbol}")
    print(f"  描述: {flow.description}")

持仓查询

查询股票持仓

# 查询所有股票持仓
positions = ctx.stock_positions()
print(f"渠道: {positions.channels}")  # 持仓渠道列表

for pos in positions.positions:
    print(f"{pos.symbol}:")
    print(f"  数量: {pos.quantity}")
    print(f"  可用数量: {pos.available_quantity}")
    print(f"  成本价: {pos.cost_price}")
    print(f"  市场: {pos.market}")
    print(f"  货币: {pos.currency}")

# 查询指定标的
positions = ctx.stock_positions(symbols=["700.HK", "AAPL.US"])

查询基金持仓

# 查询基金持仓
fund_positions = ctx.fund_positions()

for pos in fund_positions.positions:
    print(f"{pos.symbol}:")
    print(f"  当前净值: {pos.current_net_asset_value}")
    print(f"  持有份额: {pos.holding_units}")

保证金和购买力

查询保证金比例

# 查询保证金比例
margin = ctx.margin_ratio(symbol="700.HK")
print(f"初始保证金率: {margin.im_factor}")
print(f"维持保证金率: {margin.mm_factor}")
print(f"强平保证金率: {margin.fm_factor}")

估算最大购买数量

from longport.openapi import OrderSide, OrderType, Market
from decimal import Decimal

# 估算最大可买数量
estimate = ctx.estimate_max_purchase_quantity(
    symbol="700.HK",
    order_type=OrderType.LO,
    side=OrderSide.Buy,
    price=Decimal("50.00"),
    currency="HKD",
    order_id=None,  # 可选，修改订单时使用
    market=Market.HK
)

print(f"现金最大可买: {estimate.cash_max_qty}")
print(f"融资最大可买: {estimate.margin_max_qty}")

数据结构

市场枚举

from longport.openapi import Market

Market.Unknown  # 未知
Market.US       # 美股
Market.HK       # 港股
Market.CN       # A股
Market.SG       # 新加坡
Market.Crypto   # 加密货币

订单类型

from longport.openapi import OrderType

# 基础订单
OrderType.LO    # 限价单
OrderType.ELO   # 增强限价单
OrderType.MO    # 市价单
OrderType.AO    # 竞价单
OrderType.ALO   # 竞价限价单
OrderType.ODD   # 零股单
OrderType.SLO   # 特殊限价单

# 条件订单
OrderType.LIT   # 限价触发单
OrderType.MIT   # 市价触发单

# 跟踪订单
OrderType.TSLPAMT  # 跟踪限价触发（金额）
OrderType.TSLPPCT  # 跟踪限价触发（百分比）
OrderType.TSMAMT   # 跟踪市价触发（金额）
OrderType.TSMPCT   # 跟踪市价触发（百分比）

订单状态

from longport.openapi import OrderStatus

OrderStatus.Unknown              # 未知
OrderStatus.NotReported          # 未报
OrderStatus.ReplacedNotReported  # 未报（替换订单）
OrderStatus.ProtectedNotReported # 未报（保护订单）
OrderStatus.VarietiesNotReported # 未报（条件订单）
OrderStatus.Filled               # 已成交
OrderStatus.WaitToNew            # 等待新订单
OrderStatus.New                  # 新订单
OrderStatus.WaitToReplace        # 等待替换
OrderStatus.PendingReplace       # 待替换
OrderStatus.Replaced             # 已替换
OrderStatus.PartialFilled        # 部分成交
OrderStatus.WaitToCancel         # 等待取消
OrderStatus.PendingCancel        # 待取消
OrderStatus.Rejected             # 已拒绝
OrderStatus.Canceled             # 已取消
OrderStatus.Expired              # 已过期
OrderStatus.PartialWithdrawal    # 部分撤回

买卖方向

from longport.openapi import OrderSide

OrderSide.Buy   # 买入
OrderSide.Sell  # 卖出

时间有效期

from longport.openapi import TimeInForceType

TimeInForceType.Day             # 当日有效
TimeInForceType.GoodTilCanceled # 撤销前有效
TimeInForceType.GoodTilDate     # 指定日期前有效

盘后交易

from longport.openapi import OutsideRTH

OutsideRTH.RTHOnly   # 仅常规交易时段
OutsideRTH.AnyTime   # 任意时间
OutsideRTH.Overnight # 夜盘

订阅类型

from longport.openapi import SubType

SubType.Quote   # 行情
SubType.Depth   # 深度
SubType.Brokers # 经纪商
SubType.Trade   # 逐笔交易

K 线周期

from longport.openapi import Period

# 分钟级别
Period.Min_1    # 1分钟
Period.Min_2    # 2分钟
Period.Min_3    # 3分钟
Period.Min_5    # 5分钟
Period.Min_10   # 10分钟
Period.Min_15   # 15分钟
Period.Min_20   # 20分钟
Period.Min_30   # 30分钟
Period.Min_45   # 45分钟
Period.Min_60   # 60分钟
Period.Min_120  # 2小时
Period.Min_180  # 3小时
Period.Min_240  # 4小时

# 日线及以上
Period.Day      # 日线
Period.Week     # 周线
Period.Month    # 月线
Period.Quarter  # 季线
Period.Year     # 年线

复权类型

from longport.openapi import AdjustType

AdjustType.NoAdjust      # 不复权
AdjustType.ForwardAdjust # 前复权

交易时段

from longport.openapi import TradeSessions

TradeSessions.Intraday  # 日内交易
TradeSessions.All       # 全部交易时段

语言

from longport.openapi import Language

Language.ZH_CN  # 简体中文
Language.ZH_HK  # 繁体中文
Language.EN     # 英文

订单标签

from longport.openapi import OrderTag

OrderTag.Unknown          # 未知
OrderTag.Normal           # 普通订单
OrderTag.LongTerm         # 长期订单
OrderTag.Grey             # 暗盘订单
OrderTag.MarginCall       # 强制卖出
OrderTag.Offline          # OTC（场外交易）
OrderTag.Creditor         # 期权行权多头
OrderTag.Debtor           # 期权行权空头
OrderTag.NonExercise      # 放弃期权行权
OrderTag.AllocatedSub     # 交易分配

触发状态

from longport.openapi import TriggerStatus

TriggerStatus.Unknown     # 未知
TriggerStatus.Deactive    # 未激活
TriggerStatus.Active      # 已激活
TriggerStatus.Released    # 已触发

免佣状态

from longport.openapi import CommissionFreeStatus

CommissionFreeStatus.Unknown    # 未知
CommissionFreeStatus.None_      # 无
CommissionFreeStatus.Calculated # 待计算免佣金额
CommissionFreeStatus.Pending    # 待免佣
CommissionFreeStatus.Ready      # 已免佣

扣减状态

from longport.openapi import DeductionStatus

DeductionStatus.Unknown   # 未知
DeductionStatus.None_     # 无
DeductionStatus.NoData    # 已结算无数据
DeductionStatus.Pending   # 已结算待发放
DeductionStatus.Done      # 已结算已发放

费用分类代码

from longport.openapi import ChargeCategoryCode

ChargeCategoryCode.Unknown  # 未知
ChargeCategoryCode.Broker   # 经纪商费用
ChargeCategoryCode.Third    # 第三方费用

余额类型

from longport.openapi import BalanceType

BalanceType.Unknown  # 未知
BalanceType.Cash     # 现金
BalanceType.Stock    # 股票
BalanceType.Fund     # 基金

资金流向

from longport.openapi import CashFlowDirection

CashFlowDirection.Unknown  # 未知
CashFlowDirection.Out      # 流出
CashFlowDirection.In       # 流入

交易时段

from longport.openapi import TradeSession

TradeSession.Normal       # 正常交易时段
TradeSession.PreTrading   # 盘前交易
TradeSession.PostTrading  # 盘后交易

交易状态

from longport.openapi import TradeStatus

TradeStatus.Normal              # 正常交易
TradeStatus.Halted              # 停牌
TradeStatus.Delisted            # 退市
TradeStatus.Fuse                # 熔断
TradeStatus.PrepareList         # 待上市
TradeStatus.CodeMoved           # 代码迁移
TradeStatus.ToBeOpened          # 待开盘
TradeStatus.SplitStockHalts     # 拆股暂停
TradeStatus.Expired             # 已过期
TradeStatus.WarrantPrepareList  # 窝轮待上市
TradeStatus.Suspend             # 暂停交易

权证类型

from longport.openapi import WarrantType

WarrantType.Unknown
WarrantType.Call      # 认购证
WarrantType.Put       # 认沽证
WarrantType.Bull      # 牛证
WarrantType.Bear      # 熊证
WarrantType.Inline    # 界内证

期权类型

from longport.openapi import OptionType

OptionType.Unknown
OptionType.American  # 美式期权
OptionType.Europe    # 欧式期权

期权方向

from longport.openapi import OptionDirection

OptionDirection.Unknown
OptionDirection.Put   # 看跌期权
OptionDirection.Call  # 看涨期权

衍生品类型

from longport.openapi import DerivativeType

DerivativeType.Option   # 美股期权
DerivativeType.Warrant  # 港股窝轮

错误处理

OpenApiException

SDK 统一使用 OpenApiException 处理所有错误。

from longport.openapi import OpenApiException

try:
    resp = ctx.submit_order(...)
except OpenApiException as e:
    print(f"错误代码: {e.code}")
    print(f"错误消息: {e.message}")

    # 根据错误码处理
    if e.code == 401:
        print("认证失败")
    elif e.code == 429:
        print("请求过于频繁")
    elif e.code == 500:
        print("服务器错误")

常见错误处理模式

import time
from longport.openapi import OpenApiException

def submit_order_with_retry(ctx, max_retries=3, **order_params):
    """带重试的下单函数"""
    for attempt in range(max_retries):
        try:
            return ctx.submit_order(**order_params)
        except OpenApiException as e:
            if e.code == 429:  # 限流
                wait_time = 2 ** attempt  # 指数退避
                print(f"请求限流，等待 {wait_time} 秒后重试...")
                time.sleep(wait_time)
            elif e.code >= 500:  # 服务器错误
                print(f"服务器错误，重试 {attempt + 1}/{max_retries}")
                time.sleep(1)
            else:
                raise  # 其他错误不重试
    raise Exception("达到最大重试次数")

# 使用
try:
    resp = submit_order_with_retry(
        ctx,
        side=OrderSide.Buy,
        symbol="700.HK",
        order_type=OrderType.LO,
        submitted_quantity=Decimal("200"),
        submitted_price=Decimal("50.00"),
        time_in_force=TimeInForceType.Day
    )
except Exception as e:
    print(f"下单失败: {e}")

设计模式

1. Context 模式

SDK 使用 Context 对象管理连接和状态，所有 API 调用都通过 Context 进行。

# 行情 Context
quote_ctx = QuoteContext(config)

# 交易 Context
trade_ctx = TradeContext(config)

# Context 自动管理 WebSocket 连接
# 无需手动连接/断开

2. 回调模式

所有推送数据都通过回调函数处理，实现异步数据流。

def on_quote(symbol: str, event: PushQuote):
    # 处理行情数据
    pass

ctx.set_on_quote(on_quote)
ctx.subscribe(["700.HK"], [SubType.Quote])

3. 订阅模式

订阅后持续接收推送数据，无需轮询。

# 订阅
ctx.subscribe(symbols, sub_types, is_first_push=True)

# 取消订阅
ctx.unsubscribe(symbols, sub_types)

# 查询当前订阅
subscriptions = ctx.subscriptions()

4. 本地缓存模式

订阅后的数据会缓存在本地，可快速获取最新数据。

# 订阅后，数据缓存在本地
ctx.subscribe(["700.HK"], [SubType.Quote])

# 随时从本地缓存获取
realtime_quote = ctx.realtime_quote(["700.HK"])

最佳实践

1. 连接管理

from contextlib import contextmanager

@contextmanager
def create_quote_context():
    """使用上下文管理器确保资源释放"""
    config = Config.from_env()
    ctx = QuoteContext(config)
    try:
        yield ctx
    finally:
        # 清理资源（如果需要）
        pass

# 使用
with create_quote_context() as ctx:
    quotes = ctx.quote(["700.HK"])
    # Context 会自动管理

2. 订阅管理

class QuoteSubscriber:
    """行情订阅管理器"""

    def __init__(self, config):
        self.ctx = QuoteContext(config)
        self.subscribed_symbols = set()

    def subscribe_quote(self, symbols: list, callback):
        """订阅行情"""
        self.ctx.set_on_quote(callback)
        self.ctx.subscribe(symbols, [SubType.Quote], is_first_push=True)
        self.subscribed_symbols.update(symbols)

    def unsubscribe_all(self):
        """取消所有订阅"""
        if self.subscribed_symbols:
            self.ctx.unsubscribe(
                list(self.subscribed_symbols),
                [SubType.Quote]
            )
            self.subscribed_symbols.clear()

    def get_realtime_quotes(self):
        """获取所有订阅标的的实时行情"""
        if self.subscribed_symbols:
            return self.ctx.realtime_quote(list(self.subscribed_symbols))
        return []

# 使用
subscriber = QuoteSubscriber(Config.from_env())
subscriber.subscribe_quote(["700.HK", "AAPL.US"], on_quote)
quotes = subscriber.get_realtime_quotes()

3. 交易策略封装

from decimal import Decimal
from typing import Optional

class TradingStrategy:
    """交易策略基类"""

    def __init__(self, config):
        self.ctx = TradeContext(config)

    def market_buy(self, symbol: str, quantity: Decimal,
                   remark: Optional[str] = None):
        """市价买入"""
        return self.ctx.submit_order(
            side=OrderSide.Buy,
            symbol=symbol,
            order_type=OrderType.MO,
            submitted_quantity=quantity,
            time_in_force=TimeInForceType.Day,
            remark=remark or f"市价买入 {symbol}"
        )

    def limit_buy(self, symbol: str, quantity: Decimal,
                  price: Decimal, remark: Optional[str] = None):
        """限价买入"""
        return self.ctx.submit_order(
            side=OrderSide.Buy,
            symbol=symbol,
            order_type=OrderType.LO,
            submitted_quantity=quantity,
            submitted_price=price,
            time_in_force=TimeInForceType.Day,
            remark=remark or f"限价买入 {symbol}"
        )

    def stop_loss(self, symbol: str, quantity: Decimal,
                  trigger_price: Decimal):
        """止损单"""
        return self.ctx.submit_order(
            side=OrderSide.Sell,
            symbol=symbol,
            order_type=OrderType.MIT,
            submitted_quantity=quantity,
            trigger_price=trigger_price,
            time_in_force=TimeInForceType.GoodTilCanceled,
            remark=f"止损 {symbol} @{trigger_price}"
        )

    def trailing_stop(self, symbol: str, quantity: Decimal,
                     trigger_price: Decimal, trailing_percent: Decimal):
        """跟踪止损"""
        return self.ctx.submit_order(
            side=OrderSide.Sell,
            symbol=symbol,
            order_type=OrderType.TSMPCT,
            submitted_quantity=quantity,
            trigger_price=trigger_price,
            trailing_percent=trailing_percent,
            time_in_force=TimeInForceType.GoodTilCanceled,
            remark=f"跟踪止损 {symbol}"
        )

# 使用
strategy = TradingStrategy(Config.from_env())
resp = strategy.limit_buy("700.HK", Decimal("200"), Decimal("50.00"))
print(f"订单ID: {resp.order_id}")

4. 批量操作

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List

class BatchTrader:
    """批量交易执行器"""

    def __init__(self, config, max_workers=5):
        self.ctx = TradeContext(config)
        self.max_workers = max_workers

    def batch_submit_orders(self, orders: List[dict]) -> List[dict]:
        """批量提交订单"""
        results = []

        with ThreadPoolExecutor(max_workers=self.max_workers) as executor:
            futures = {
                executor.submit(self.ctx.submit_order, **order): order
                for order in orders
            }

            for future in as_completed(futures):
                order = futures[future]
                try:
                    resp = future.result()
                    results.append({
                        "success": True,
                        "order_id": resp.order_id,
                        "symbol": order["symbol"]
                    })
                except Exception as e:
                    results.append({
                        "success": False,
                        "error": str(e),
                        "symbol": order["symbol"]
                    })

        return results

# 使用
trader = BatchTrader(Config.from_env())
orders = [
    {
        "side": OrderSide.Buy,
        "symbol": "700.HK",
        "order_type": OrderType.LO,
        "submitted_quantity": Decimal("100"),
        "submitted_price": Decimal("50.00"),
        "time_in_force": TimeInForceType.Day
    },
    {
        "side": OrderSide.Buy,
        "symbol": "AAPL.US",
        "order_type": OrderType.LO,
        "submitted_quantity": Decimal("10"),
        "submitted_price": Decimal("150.00"),
        "time_in_force": TimeInForceType.Day
    }
]

results = trader.batch_submit_orders(orders)
for result in results:
    if result["success"]:
        print(f"{result['symbol']} 下单成功: {result['order_id']}")
    else:
        print(f"{result['symbol']} 下单失败: {result['error']}")

5. 数据缓存和持久化

import json
from datetime import datetime
from pathlib import Path

class DataCache:
    """数据缓存管理器"""

    def __init__(self, cache_dir="./cache"):
        self.cache_dir = Path(cache_dir)
        self.cache_dir.mkdir(exist_ok=True)

    def save_quotes(self, quotes: list):
        """保存行情数据"""
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        filename = self.cache_dir / f"quotes_{timestamp}.json"

        data = [
            {
                "symbol": q.symbol,
                "last_done": str(q.last_done),
                "open": str(q.open),
                "high": str(q.high),
                "low": str(q.low),
                "volume": q.volume,
                "timestamp": q.timestamp.isoformat()
            }
            for q in quotes
        ]

        with open(filename, "w") as f:
            json.dump(data, f, indent=2)

        return filename

    def save_candlesticks(self, symbol: str, period: str,
                         candlesticks: list):
        """保存 K 线数据"""
        filename = self.cache_dir / f"{symbol}_{period}.json"

        data = [
            {
                "timestamp": c.timestamp.isoformat(),
                "open": str(c.open),
                "high": str(c.high),
                "low": str(c.low),
                "close": str(c.close),
                "volume": c.volume,
                "turnover": str(c.turnover)
            }
            for c in candlesticks
        ]

        with open(filename, "w") as f:
            json.dump(data, f, indent=2)

# 使用
cache = DataCache()
quotes = ctx.quote(["700.HK", "AAPL.US"])
cache.save_quotes(quotes)

6. 高频交易配置

class HighFrequencyTrader:
    """高频交易配置"""

    def __init__(self, config):
        self.ctx = TradeContext(config)

    def hk_scalping_order(self, symbol: str, quantity: Decimal,
                         price: Decimal):
        """港股日内高频单（使用增强限价单）"""
        return self.ctx.submit_order(
            side=OrderSide.Buy,
            symbol=symbol,
            order_type=OrderType.ELO,  # 增强限价单
            submitted_quantity=quantity,
            submitted_price=price,
            time_in_force=TimeInForceType.Day,  # 当日有效
            outside_rth=OutsideRTH.RTHOnly,
            remark="高频日内交易"
        )

    def us_momentum_order(self, symbol: str, quantity: Decimal,
                         trigger_price: Decimal, limit_offset: Decimal):
        """美股突破策略"""
        return self.ctx.submit_order(
            side=OrderSide.Buy,
            symbol=symbol,
            order_type=OrderType.LIT,  # 限价触发单
            submitted_quantity=quantity,
            trigger_price=trigger_price,
            submitted_price=trigger_price + limit_offset,
            time_in_force=TimeInForceType.Day,
            outside_rth=OutsideRTH.AnyTime,  # 支持盘后
            remark="突破买入"
        )

# 使用
hft = HighFrequencyTrader(Config.from_env())
resp = hft.hk_scalping_order("700.HK", Decimal("200"), Decimal("50.00"))

7. 日志和监控

import logging
from datetime import datetime

class TradingLogger:
    """交易日志管理器"""

    def __init__(self, log_file="trading.log"):
        self.logger = logging.getLogger("LongPort")
        self.logger.setLevel(logging.INFO)

        # 文件处理器
        fh = logging.FileHandler(log_file)
        fh.setLevel(logging.INFO)

        # 控制台处理器
        ch = logging.StreamHandler()
        ch.setLevel(logging.WARNING)

        # 格式化
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        fh.setFormatter(formatter)
        ch.setFormatter(formatter)

        self.logger.addHandler(fh)
        self.logger.addHandler(ch)

    def log_order(self, action: str, order_params: dict,
                  result: Optional[str] = None):
        """记录订单操作"""
        msg = f"{action} - {order_params['symbol']} - "
        msg += f"{order_params['side']} {order_params['submitted_quantity']}"
        if result:
            msg += f" - Result: {result}"
        self.logger.info(msg)

    def log_error(self, action: str, error: Exception):
        """记录错误"""
        self.logger.error(f"{action} failed: {str(error)}")

# 使用
logger = TradingLogger()

try:
    resp = ctx.submit_order(
        side=OrderSide.Buy,
        symbol="700.HK",
        order_type=OrderType.LO,
        submitted_quantity=Decimal("200"),
        submitted_price=Decimal("50.00"),
        time_in_force=TimeInForceType.Day
    )
    logger.log_order("SUBMIT_ORDER", {
        "symbol": "700.HK",
        "side": OrderSide.Buy,
        "submitted_quantity": Decimal("200")
    }, result=resp.order_id)
except OpenApiException as e:
    logger.log_error("SUBMIT_ORDER", e)

8. 配置管理

from typing import Dict, Any
import yaml

class TradingConfig:
    """交易配置管理器"""

    def __init__(self, config_file="trading_config.yaml"):
        self.config_file = config_file
        self.config = self.load_config()

    def load_config(self) -> Dict[str, Any]:
        """加载配置文件"""
        try:
            with open(self.config_file, 'r') as f:
                return yaml.safe_load(f)
        except FileNotFoundError:
            return self.default_config()

    def default_config(self) -> Dict[str, Any]:
        """默认配置"""
        return {
            "trading": {
                "default_time_in_force": "Day",
                "default_outside_rth": "RTHOnly",
                "max_position_size": 10000,
                "max_daily_trades": 100
            },
            "risk": {
                "max_loss_per_trade": 0.02,  # 2%
                "max_portfolio_loss": 0.10,   # 10%
                "stop_loss_enabled": True
            },
            "markets": {
                "HK": {
                    "preferred_order_type": "ELO",
                    "lot_size_check": True
                },
                "US": {
                    "preferred_order_type": "LO",
                    "allow_pre_market": True,
                    "allow_after_market": True
                }
            }
        }

    def get(self, key: str, default=None):
        """获取配置项"""
        keys = key.split('.')
        value = self.config
        for k in keys:
            value = value.get(k)
            if value is None:
                return default
        return value

# 使用
config = TradingConfig()
max_position = config.get("trading.max_position_size")
hk_order_type = config.get("markets.HK.preferred_order_type")

完整示例

简单量化策略示例

"""
简单的移动平均线交叉策略
"""
from decimal import Decimal
from time import sleep
from longport.openapi import (
    Config, QuoteContext, TradeContext,
    Period, AdjustType, TradeSessions,
    OrderSide, OrderType, TimeInForceType
)

class MAStrategy:
    """移动平均线策略"""

    def __init__(self):
        config = Config.from_env()
        self.quote_ctx = QuoteContext(config)
        self.trade_ctx = TradeContext(config)
        self.position = {}  # 持仓

    def calculate_ma(self, prices: list, period: int) -> float:
        """计算移动平均"""
        if len(prices) < period:
            return 0
        return sum(prices[-period:]) / period

    def get_prices(self, symbol: str, count: int = 50) -> list:
        """获取历史价格"""
        candlesticks = self.quote_ctx.candlesticks(
            symbol=symbol,
            period=Period.Day,
            count=count,
            adjust_type=AdjustType.ForwardAdjust,
            trade_sessions=TradeSessions.Intraday
        )
        return [float(c.close) for c in candlesticks]

    def check_signal(self, symbol: str):
        """检查交易信号"""
        prices = self.get_prices(symbol)

        # 计算 MA5 和 MA20
        ma5 = self.calculate_ma(prices, 5)
        ma20 = self.calculate_ma(prices, 20)

        current_price = prices[-1]

        print(f"{symbol}: 当前价={current_price:.2f}, "
              f"MA5={ma5:.2f}, MA20={ma20:.2f}")

        # 金叉：MA5 上穿 MA20，买入信号
        if ma5 > ma20 and symbol not in self.position:
            return "BUY"

        # 死叉：MA5 下穿 MA20，卖出信号
        elif ma5 < ma20 and symbol in self.position:
            return "SELL"

        return "HOLD"

    def execute_trade(self, symbol: str, signal: str):
        """执行交易"""
        if signal == "BUY":
            try:
                resp = self.trade_ctx.submit_order(
                    side=OrderSide.Buy,
                    symbol=symbol,
                    order_type=OrderType.MO,
                    submitted_quantity=Decimal("100"),
                    time_in_force=TimeInForceType.Day,
                    remark="MA策略买入"
                )
                self.position[symbol] = resp.order_id
                print(f"✓ 买入 {symbol}, 订单ID: {resp.order_id}")
            except Exception as e:
                print(f"✗ 买入失败: {e}")

        elif signal == "SELL" and symbol in self.position:
            try:
                resp = self.trade_ctx.submit_order(
                    side=OrderSide.Sell,
                    symbol=symbol,
                    order_type=OrderType.MO,
                    submitted_quantity=Decimal("100"),
                    time_in_force=TimeInForceType.Day,
                    remark="MA策略卖出"
                )
                del self.position[symbol]
                print(f"✓ 卖出 {symbol}, 订单ID: {resp.order_id}")
            except Exception as e:
                print(f"✗ 卖出失败: {e}")

    def run(self, symbols: list, interval: int = 60):
        """运行策略"""
        print("策略启动...")

        while True:
            for symbol in symbols:
                signal = self.check_signal(symbol)
                if signal != "HOLD":
                    self.execute_trade(symbol, signal)

            print(f"等待 {interval} 秒...")
            sleep(interval)

# 运行策略
if __name__ == "__main__":
    strategy = MAStrategy()
    strategy.run(["700.HK", "AAPL.US"], interval=300)  # 5分钟检查一次

附录

A. 完整的环境变量列表

# 必需
LONGPORT_APP_KEY=your_app_key
LONGPORT_APP_SECRET=your_app_secret
LONGPORT_ACCESS_TOKEN=your_access_token

# 可选
LONGPORT_LANGUAGE=zh-CN
LONGPORT_HTTP_URL=https://openapi.longportapp.com
LONGPORT_QUOTE_WS_URL=wss://openapi-quote.longportapp.com
LONGPORT_TRADE_WS_URL=wss://openapi-trade.longportapp.com
LONGPORT_ENABLE_OVERNIGHT=false
LONGPORT_PUSH_CANDLESTICK_MODE=realtime
LONGPORT_PRINT_QUOTE_PACKAGES=true
LONGPORT_LOG_PATH=/path/to/log

B. 市场特定注意事项

港股（HK）

支持 T+0 交易
无涨跌停限制
推荐使用增强限价单（ELO）
交易时段：09:30-12:00, 13:00-16:00

美股（US）

支持 T+0 交易
支持盘前盘后交易
可使用 outside_rth 参数
交易时段：09:30-16:00（盘中），04:00-09:30（盘前），16:00-20:00（盘后）

A股（CN）

T+1 交易
有涨跌停限制（10%，ST为5%）
集合竞价：09:15-09:25，14:57-15:00

C. API 限流说明

LongPort API 有限流限制，请合理控制请求频率：

行情 API：每秒建议不超过 10 次
交易 API：每秒建议不超过 5 次
批量查询时使用单次查询多个标的，而非多次查询单个标的

D. 数值精度

使用 Decimal 类型处理金额和价格，避免浮点数精度问题：

from decimal import Decimal

# 正确
price = Decimal("50.00")
quantity = Decimal("200")

# 错误（避免）
price = 50.00  # float 类型
quantity = 200  # int 类型

文档版本: 1.0 更新时间: 2025-10-12 SDK 版本: longport >= 1.0.0 官方文档: https://longportapp.github.io/openapi/python/ GitHub: https://github.com/longportapp/openapi

LongPort OpenAPI Python SDK 开发指南

目录

快速开始

安装

基本使用流程

认证和配置

Config 类

类定义

创建配置

Access Token 刷新

HttpClient 类

核心接口

QuoteContext - 行情上下文

初始化

TradeContext - 交易上下文

初始化

行情 API

订阅机制

订阅行情（实时报价）

订阅深度（买卖盘）

订阅经纪商队列

订阅逐笔交易

订阅 K 线

查询订阅信息

获取证券信息

基本信息（静态数据）

实时行情

期权行情

权证行情

深度和经纪商数据

逐笔交易

K 线数据

获取最近 K 线

历史 K 线（按偏移量）

历史 K 线（按日期范围）

日内分时数据

期权链

权证筛选

市场信息

交易日历

当日交易时段

市场温度

资金流向

自选股

计算指标

实时数据（从本地缓存）

交易 API

订单提交

submit_order 方法定义

基本订单类型

条件订单

跟踪订单

时间有效期

盘后交易（美股）

订单管理

修改订单

取消订单

订单查询

查询今日订单

查询历史订单

查询订单详情

成交查询

查询今日成交

查询历史成交

订单推送通知

账户信息

查询账户余额

查询资金流水

持仓查询

查询股票持仓

查询基金持仓

保证金和购买力

查询保证金比例

估算最大购买数量

数据结构

市场枚举

订单类型

订单状态

买卖方向

时间有效期