ptrade常见问题--2常见量化问题解答

: 小果量化交易量化研究专做量化开发研究，微信xg_quant

0 人点赞了该文章 · 63 浏览

ptrade常见问题--2常见量化问题解答

风险提示：本内容为量化技术及平台操作的纯知识分享，不构成任何投资建议，不推荐任何具体证券或交易时机。部分内容可能理解包含错误，请自行核实。市场有风险，投资需谨慎

今天继续按计划给大家更新量化教程，今天更新ptrade常见问题第二期，前面更新了一个内容ptrade常见问题-常见问题解答第一篇

更新的内容计划，前面更新了算法研究--日内拐点交易算法原理研究，点赞加关注后面持续更新内容

1 策略算法原理2qmt使用教程3策略算法原理4量化研报5 ptrad使用教程6 因子分析7策略算法原理

1run_daily用法及运行时间

1、该函数用于以日为单位周期性运行指定函数，可对运行触发时间进行指定。

2、该函数只能在初始化阶段initialize函数中调用。

3、该函数可以在initialize中多次调用，以实现多个定时任务。但需要注意的是交易中定时任务线程数限制为5且累计的任务不执行，即run_daily和run_interval累计调用超过5次时，将会因堵塞导致部分定时任务不触发。

4、股票策略回测中，当回测周期为分钟时，time的取值指在09:31~11:30与13:00~15:00之间，当回测周期为日时，无论设定值是多少都只会在15:00执行；交易中不受此时间限制。

代码框架参考

# -*- coding: utf-8 -*-"""run_daily 用法示例（优化版）展示了多个定时任务的配置方式，避免与内部函数名冲突"""def initialize(context):    # 设置股票池    g.security = '6.SS'    set_universe(g.security)
    # ==================== 多个定时任务示例 ====================
    # 1. 盘前任务：9:02 运行（交易环境可用）    run_daily(context, my_before_market_open, time='09:02')
    # 2. 集合竞价任务：9:23 运行（回测分钟级可用）    run_daily(context, my_aggregate_auction_func, time='9:23')
    # 3. 盘中任务：10:00 运行（回测分钟级可用）    run_daily(context, my_mid_day_task, time='10:00')
    # 4. 尾盘任务：14:50 运行（回测分钟级可用）    run_daily(context, my_after_market_close, time='14:50')
    # 5. 盘后任务：15:10 运行（交易环境可用，回测日线级在15:00执行）    run_daily(context, my_post_market_task, time='15:10')
    log.info("定时任务设置完成，共5个任务")# ==================== 定时任务函数定义 ====================def my_before_market_open(context):    """盘前任务：9:02 执行"""    log.info(f"盘前任务执行 - {context.current_dt}")    # 获取历史数据等准备工作    history_data = get_history(10, '1d', 'close', g.security, fq='pre', include=True)    log.info(f"获取历史数据完成")def my_aggregate_auction_func(context):    """集合竞价任务：9:23 执行"""    log.info(f"集合竞价任务执行 - {context.current_dt}")    stock = g.security    # 获取最新价和涨停价    snapshot = get_snapshot(stock)    price = snapshot[stock]['last_px']    up_limit = snapshot[stock]['up_px']    # 如果最新价不小于涨停价则买入    if float(price) >= float(up_limit):        order(g.security, 100, limit_price=up_limit)        log.info(f"集合竞价买入: {stock}")def my_mid_day_task(context):    """盘中任务：10:00 执行"""    log.info(f"盘中任务执行 - {context.current_dt}")    # 检查持仓状态    positions = get_positions()    log.info(f"当前持仓: {positions}")def my_after_market_close(context):    """尾盘任务：14:50 执行"""    log.info(f"尾盘任务执行 - {context.current_dt}")    # 执行调仓操作    stock = g.security    # 获取当前价格    snapshot = get_snapshot(stock)    price = snapshot[stock]['last_px']    # 买入操作    order(g.security, 100)    log.info(f"尾盘买入: {stock}")def my_post_market_task(context):    """盘后任务：15:10 执行"""    log.info(f"盘后任务执行 - {context.current_dt}")    # 盘后处理：如逆回购、日志记录等    cash = context.portfolio.cash    log.info(f"盘后资金余额: {cash:.2f}")# ==================== 必选函数 ====================def handle_data(context, data):    """盘中处理函数（日线级别每天15:00执行一次）"""    # 日线策略中，handle_data 在15:00执行    # 这里可以放置主要的交易逻辑    pass

特别注意定时超了5个可能后面的模块拉不起来

# ❌ 错误写法：超过5个定时任务会导致部分任务不执行run_daily(context, task1, time='09:00')run_daily(context, task2, time='09:30')run_daily(context, task3, time='10:00')run_daily(context, task4, time='10:30')run_daily(context, task5, time='11:00')run_daily(context, task6, time='11:30')  # 第6个任务不会执行# ✅ 正确写法：合并任务或减少任务数量run_daily(context, combined_task, time='10:00')  # 将多个逻辑合并到一个函数

2run_interval用法及运行时间

1、该函数用于以设定时间间隔(单位为秒)周期性运行指定函数，可对运行触发时间间隔进行指定。

2、该函数只能在初始化阶段initialize函数中调用。

3、运行时间(1) PTrade1.0-QTV202501.04.000版本之前，启动当天根据启动时间+设置周期，第二天则是9：30 + 设置周期(2) PTrade1.0-QTV202501.04.000版本（含）之后，run_interval接口增加入参 interval_timer_ranges字段，支持设置运行的时间段，即支持配置24小时都能运行，例如配置：“09:15-11:30, 13:00-15:00”表示从上午9点15分到11点半和下午1点到3点的时间范围。

4、该函数可以在initialize中多次调用，以实现多个定时任务。但需要注意的是交易中定时任务线程数限制为5且累计的任务不执行，即run_daily和run_interval累计调用超过5次时，将会因堵塞导致部分定时任务不触发。

5、最小运行时间间隔seconds的设置规则：期货策略为1秒（用户设置值若小于1秒，系统仍当做1秒处理），股票等其他类型策略为3秒。

代码

# -*- coding: utf-8 -*-"""run_interval 用法示例（优化版）展示了多个定时任务的配置方式，避免与内部函数名冲突"""def initialize(context):    # 设置股票池    g.security = '60.SS'    set_universe(g.security)
    # ==================== 多个 run_interval 定时任务示例 ====================
    # 1. 每3秒运行一次（股票最小间隔）    run_interval(context, my_tick_handle, seconds=3)
    # 2. 每10秒运行一次（常规间隔）    run_interval(context, my_interval_handle, seconds=10)
    # 3. 每60秒运行一次（1分钟间隔）    run_interval(context, my_minute_handle, seconds=60)
    log.info("定时任务设置完成，共3个任务")# ==================== 定时任务函数定义 ====================def my_tick_handle(context):    """tick级别处理函数：每3秒执行一次"""    log.info(f"tick级别任务执行 - {context.current_dt}")
    stock = g.security    # 获取最新行情快照    snapshot = get_snapshot(stock)    price = snapshot[stock]['last_px']
    log.info(f"股票 {stock} 最新价: {price}")
    # 示例：简单的价格监控逻辑    if price > 0:        # 获取当前持仓        position = get_position(stock)        if position is None or position.amount == 0:            # 没有持仓，考虑买入            log.info(f"当前无持仓，监控价格: {price}")        else:            # 有持仓，监控盈亏            cost = position.cost_basis            profit_ratio = (price - cost) / cost * 100            log.info(f"持仓盈亏比例: {profit_ratio:.2f}%")def my_interval_handle(context):    """常规间隔处理函数：每10秒执行一次"""    log.info(f"常规间隔任务执行 - {context.current_dt}")
    # 获取账户信息    total_value = context.portfolio.portfolio_value    cash = context.portfolio.cash
    log.info(f"账户总资产: {total_value:.2f}, 可用资金: {cash:.2f}")
    # 获取当前持仓    positions = get_positions()    if positions:        for stock, pos in positions.items():            log.info(f"持仓 {stock}: {pos.amount}股, 成本 {pos.cost_basis:.2f}")    else:        log.info("当前无持仓")def my_minute_handle(context):    """分钟级别处理函数：每60秒执行一次"""    log.info(f"分钟级别任务执行 - {context.current_dt}")
    stock = g.security    # 获取历史数据用于技术分析    history_data = get_history = get_history(5, '1d', 'close', stock, fq='pre', include=True)
    if history_history is not None and not history_history.empty:        close_prices = history_history['close'].values        ma5 = close_prices.mean()
        # 获取当前价格        snapshot = get_snapshot(stock)        current_price = snapshot[stock]['last_px']
        log.info(f"股票 {stock} 当前价: {current_price:.2f}, 5日均线: {ma5:.2f}")
        # 示例：均线策略逻辑        if current_price > ma5:            log.info(f"价格在均线上方，趋势偏多")        else:            log.info(f"价格在均线下方，趋势偏空")# ==================== 必选函数 ====================def handle_data(context, data):    """盘中处理函数（日线级别每天15:00执行一次）"""    # 日线策略中，handle_data 在15:00执行    # 这里可以放置主要的交易逻辑    pass

注意默认的数量5个

# ❌ 错误写法：超过5个定时任务会导致部分任务不执行run_interval(context, my_task1, seconds=3)run_interval(context, my_task2, seconds=5)run_interval(context, my_task3, seconds=10)run_interval(context, my_task4, seconds=30)run_interval(context, my_task5, seconds=60)run_interval(context, my_task6, seconds=120)  # 第6个任务不会执行# ✅：合并任务或减少任务数量run_interval(context, my_combined_task, seconds=10)  # 将多个逻辑合并到一个函数

最小间隔设置

# 股票策略：最小间隔为3秒run_interval(context, my_task, seconds=3)  # ✅ 有效run_interval(context, my_task, seconds=1)  # ❌ 会被自动调整为3秒# 期货策略：最小间隔为1秒run_interval(context, my_task, seconds=1)  # ✅ 有效run_interval(context, my_task, seconds=0.5)  # ❌ 会被自动调整为1秒

多线程并行注意事项

def initialize(context):    # 多个 run_interval 会以多个线程并行运行    run_interval(context, my_task_a, seconds=3)    run_interval(context, my_task_b, seconds=5)
    # ⚠️ 小心不同线程之间的逻辑关联处理    # 例如：不要在多个线程中同时修改同一个全局变量而不加锁def my_task_a(context):    # 线程A：读取数据    g.shared_data = get_snapshot(g.security)  # 可能被线程B同时修改def my_task_b(context):    # 线程B：写入数据    g.shared_data = get_history(10, '1d', 'close', g.security)  # 可能覆盖线程A的数据

3量化策略报错：无法获取到000300.SH对应的数据，设置基准失败。

1、量化中上海市场代码的尾缀为SS，不是SH，客户可将000300.SH改成000300.SS；

2、代码不确定尾缀的可以在量化回测策略中，输入代码即可联想到后缀，选择即可。

4量化的历史行情数据，支持什么周期的？

frequency：K线周期，现有支持1分钟线（1m）、5分钟线（5m）、15分钟线（15m）、30分钟线（30m）、60分钟线(60m)、120分钟线（120m）、日线（1d）、周线（1w/weekly）、月线（mo/monthly）、季度线（1q/quarter）和年线（1y/yearly)频率的数据。

5量化get_price/get_history接口能获取多久之前的数据。

能获取2005年后的数据，具体以客户环境的历史数据为准，一般生产环境能获取2005年之后的数据

6get_history函数include参数说明

get_history当日数据时include参数如果是true，就取行情服务器最新一根k线；如果是false，就是取最新一根k线的上一根。

7一分钟线与多分钟周期数据get_history 取值逻辑

1、逻辑都是：比如9:47:40秒获取，include = True, 分钟线是获取到094700的数据，五分钟线实际获取到的是9.47时候的五分钟线，此时的时间戳是9.50; 2、分钟线将最新的一根切除了，五分钟线没有切除，也就是五分钟线是实时请求到行情服务器最新的结果; 3、其他多分钟周期的，跟五分钟逻辑一样