QMT框架 API 文档

MiniQMT量化开发框架核心组件API参考

QMTEngine

整个MiniQMT量化开发框架的核心引擎，负责统一管理策略、事件和频率调度。

初始化方法

__init__(session_id: Optional[int] = None)

功能：初始化QMT引擎，设置会话ID和核心组件

参数：

  • session_id (Optional[int]): 会话ID，如果为None则使用当前时间戳

返回值：无

连接管理

connect_qmt() → bool

功能：连接QMT交易系统

参数：无

返回值：bool - 是否连接成功

disconnect_qmt()

功能：断开QMT连接

参数：无

返回值：无

策略管理

add_strategy(strategy: StrategyBase) → bool

功能：添加策略到引擎

参数：

  • strategy (StrategyBase): 策略实例

返回值：bool - 是否添加成功

remove_strategy(strategy_name: str) → bool

功能：移除策略

参数：

  • strategy_name (str): 策略名称

返回值：bool - 是否移除成功

数据订阅

subscribe_stock(stock_code: str)

功能：订阅股票数据

参数：

  • stock_code (str): 股票代码

返回值：无

unsubscribe_stock(stock_code: str)

功能：取消订阅股票数据

参数：

  • stock_code (str): 股票代码

返回值：无

get_stock_price(stock_code: str) → Optional[float]

功能：获取股票最新价格

参数：

  • stock_code (str): 股票代码

返回值：Optional[float] - 股票价格，如果获取失败返回None

引擎控制

start()

功能：启动引擎和所有策略

参数：无

返回值：无

stop()

功能：停止引擎和所有策略

参数：无

返回值：无

get_engine_stats() → Dict[str, Any]

功能：获取引擎统计信息

参数：无

返回值：Dict[str, Any] - 包含引擎状态、策略数量、订阅股票等信息的字典

EventManager

负责管理和分发各种市场事件，实现事件驱动的策略架构。

事件订阅

subscribe(event_type: EventType, handler: Callable[[Event], None])

功能：订阅特定类型的事件

参数：

  • event_type (EventType): 事件类型
  • handler (Callable[[Event], None]): 事件处理函数

返回值：无

unsubscribe(event_type: EventType, handler: Callable[[Event], None])

功能：取消订阅特定类型的事件

参数：

  • event_type (EventType): 事件类型
  • handler (Callable[[Event], None]): 事件处理函数

返回值：无

事件发布

publish(event_type: EventType, data: Any = None)

功能：发布事件到所有订阅者

参数：

  • event_type (EventType): 事件类型
  • data (Any): 事件数据，默认为None

返回值：无

publish_event(event: Event)

功能：发布事件对象到所有订阅者

参数：

  • event (Event): 事件对象

返回值：无

管理器控制

start()

功能：启动事件管理器

参数：无

返回值：无

stop()

功能：停止事件管理器

参数：无

返回值：无

is_market_time() → bool

功能：判断是否在交易时间内

参数：无

返回值：bool - 是否在交易时间内

clear_handlers()

功能：清除所有事件处理器

参数：无

返回值：无

get_event_stats() → Dict[str, int]

功能：获取事件统计信息

参数：无

返回值：Dict[str, int] - 包含各类事件处理次数的字典

事件类型 (EventType)

MARKET_OPEN MARKET_CLOSE TICK_DATA BAR_DATA ORDER_UPDATE POSITION_UPDATE

FrequencyManager

负责管理不同运行频率的任务调度，支持从tick级到月级的多种频率。

任务管理

add_task(task_id: str, frequency: RunFrequency, callback: Callable) → str

功能：添加定时任务

参数：

  • task_id (str): 任务唯一标识符
  • frequency (RunFrequency): 运行频率
  • callback (Callable): 任务回调函数

返回值：str - 任务ID

remove_task(task_id: str) → bool

功能：移除指定任务

参数：

  • task_id (str): 任务ID

返回值：bool - 是否移除成功

enable_task(task_id: str) → bool

功能：启用任务

参数：

  • task_id (str): 任务ID

返回值：bool - 是否启用成功

disable_task(task_id: str) → bool

功能：禁用任务

参数：

  • task_id (str): 任务ID

返回值：bool - 是否禁用成功

execute_frequency_tasks(frequency: RunFrequency)

功能：执行指定频率的所有任务

参数：

  • frequency (RunFrequency): 运行频率

返回值：无

execute_tick_tasks(tick_data: Dict[str, Any])

功能：执行tick级任务

参数：

  • tick_data (Dict[str, Any]): tick数据

返回值：无

管理器控制

start()

功能：启动频率管理器

参数：无

返回值：无

stop()

功能：停止频率管理器

参数：无

返回值：无

get_task_stats() → Dict[str, Any]

功能：获取任务统计信息

参数：无

返回值：Dict[str, Any] - 包含任务数量、执行次数等统计信息的字典

运行频率 (RunFrequency)

TICK MINUTE HOUR DAY WEEK MONTH

StrategyBase

策略开发的基础抽象类，提供完整的策略生命周期管理和事件处理框架。

初始化方法

__init__(strategy_name: str, run_frequency: RunFrequency = RunFrequency.MINUTE)

功能：初始化策略基类

参数：

  • strategy_name (str): 策略名称
  • run_frequency (RunFrequency): 运行频率，默认为分钟级

返回值：无

生命周期方法

initialize()

功能：策略初始化，在策略启动前调用

参数：无

返回值：无

start()

功能：启动策略

参数：无

返回值：无

stop()

功能：停止策略

参数：无

返回值：无

on_initialize() *必须实现

功能：策略初始化回调（抽象方法）

参数：无

返回值：无

on_start()

功能：策略启动回调

参数：无

返回值：无

on_stop()

功能：策略停止回调

参数：无

返回值：无

on_market_open()

功能：开盘事件回调

参数：无

返回值：无

on_market_close()

功能：收盘事件回调

参数：无

返回值：无

on_tick(tick_data: Dict[str, Any])

功能：tick数据事件回调

参数：

  • tick_data (Dict[str, Any]): tick数据字典

返回值：无

on_bar(bar_data: Dict[str, Any], frequency: str)

功能：K线数据事件回调

参数：

  • bar_data (Dict[str, Any]): K线数据字典
  • frequency (str): K线频率

返回值：无

工具方法

log_info(message: str)

功能：记录信息级别日志

参数：

  • message (str): 日志消息

返回值：无

log_warning(message: str)

功能：记录警告级别日志

参数：

  • message (str): 日志消息

返回值：无

log_error(message: str)

功能：记录错误级别日志

参数：

  • message (str): 日志消息

返回值：无

set_context(key: str, value: Any)

功能：设置策略上下文变量

参数：

  • key (str): 变量键名
  • value (Any): 变量值

返回值：无

get_context(key: str, default: Any = None) → Any

功能：获取策略上下文变量

参数：

  • key (str): 变量键名
  • default (Any): 默认值，默认为None

返回值：Any - 变量值或默认值

schedule_task(execute_time: time, callback: Callable, description: str = "") → str

功能：添加定时任务

参数：

  • execute_time (time): 执行时间
  • callback (Callable): 回调函数
  • description (str): 任务描述，默认为空字符串

返回值：str - 任务ID

FrequencyTask

频率任务类，封装了单个定时任务的执行逻辑和状态管理。

初始化方法

__init__(task_id: str, frequency: RunFrequency, callback: Callable, enabled: bool = True)

功能：初始化频率任务

参数：

  • task_id (str): 任务唯一标识符
  • frequency (RunFrequency): 任务运行频率
  • callback (Callable): 任务回调函数
  • enabled (bool): 是否启用任务，默认为True

返回值：无

任务控制方法

should_run(current_time: datetime) → bool

功能：判断任务是否应该在当前时间执行

参数：

  • current_time (datetime): 当前时间

返回值：bool - 是否应该执行

execute(*args, **kwargs)

功能：执行任务回调函数

参数：

  • *args: 位置参数
  • **kwargs: 关键字参数

返回值：无

get_stats() → Dict[str, Any]

功能：获取任务统计信息

参数：无

返回值：Dict[str, Any] - 包含执行次数、错误次数、最后执行时间等信息的字典

enable()

功能：启用任务

参数：无

返回值：无

disable()

功能：禁用任务

参数：无

返回值：无

reset_stats()

功能：重置任务统计信息

参数：无

返回值：无

任务属性

task_id: str
任务唯一标识符
frequency: RunFrequency
任务运行频率
enabled: bool
任务启用状态
run_count: int
任务执行次数统计
error_count: int
任务执行错误次数统计
last_run_time: Optional[datetime]
最后一次执行时间
next_run_time: Optional[datetime]
下次预计执行时间

MovingAverageStrategy

移动平均线策略示例，展示如何继承StrategyBase实现具体的交易策略。

初始化方法

__init__(stock_code: str, short_window: int = 5, long_window: int = 20)

功能：初始化移动平均线策略

参数：

  • stock_code (str): 交易股票代码
  • short_window (int): 短期移动平均窗口，默认为5
  • long_window (int): 长期移动平均窗口，默认为20

返回值：无

策略实现方法

on_initialize()

功能：初始化策略参数和数据结构

参数：无

返回值：无

on_bar(bar_data: Dict[str, Any], frequency: str)

功能：处理K线数据，计算移动平均线并生成交易信号

参数：

  • bar_data (Dict[str, Any]): K线数据字典
  • frequency (str): K线频率

返回值：无

calculate_moving_average(prices: List[float], window: int) → Optional[float]

功能：计算移动平均线

参数：

  • prices (List[float]): 价格列表
  • window (int): 移动平均窗口大小

返回值：Optional[float] - 移动平均值，如果数据不足返回None

generate_signal(short_ma: float, long_ma: float) → str

功能：根据移动平均线生成交易信号

参数：

  • short_ma (float): 短期移动平均值
  • long_ma (float): 长期移动平均值

返回值：str - 交易信号（"BUY", "SELL", "HOLD"）

log_performance_summary()

功能：记录策略性能摘要

参数：无

返回值：无

策略属性

stock_code: str
交易股票代码
short_window: int
短期移动平均窗口大小
long_window: int
长期移动平均窗口大小
prices: List[float]
价格历史数据列表
position: str
当前持仓状态（"NONE", "LONG", "SHORT"）

TickLevelStrategy

Tick级策略示例，展示如何处理高频tick数据进行交易决策。

初始化方法

__init__(stock_code: str, tick_threshold: float = 0.01)

功能：初始化Tick级策略

参数：

  • stock_code (str): 交易股票代码
  • tick_threshold (float): tick价格变动阈值，默认为0.01

返回值：无

策略实现方法

on_tick(tick_data: Dict[str, Any])

功能：处理tick数据并执行高频交易逻辑

参数：

  • tick_data (Dict[str, Any]): tick数据字典

返回值：无

analyze_tick_pattern(current_price: float, volume: int) → str

功能：分析tick模式并生成交易信号

参数：

  • current_price (float): 当前价格
  • volume (int): 成交量

返回值：str - 交易信号