2常见量化问题解答

落花忆流年 · 发表于 1 小时前

今天继续按计划给大家更新量化教程，今天更新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、其他多分钟周期的，跟五分钟逻辑一样

8get_history函数取五分钟线，include使用true和false有什么区别？

1、使用true时，会包含最新生成的五分钟线，例如10.35.33执行，会获取到10.40.00这一根五分钟线（行情实时生成变化，在某一分钟内是一个定值）;

2、若是false时，则在10.35.33执行的语句只能获取到10.35.00着一根五分钟线，为已经结束的时间内的五分钟线，不再会变化。

9 9点30分46秒调用 get_history(include=True)没有取到当日最新的数据。

此为正常现象，需要到9点31之后才可取到当日最新的数据，对于分钟k线，include=True取到的是最新的一根完整K线，对于日线来说，9点31分前的取到的是一上一交易日的日线，9点31分后取到的是当前交易日最新的日线。

10 量化中获取很多代码的收盘价有误差？

东财软件/同花顺/通达信默认开启复权，量化get_price、get_history接口fq-数据复权选项不设置默认不开启，所以客户对比会有误差。客户可将两个软件都调整为复权或者不复权再进行对比，正常。

更多的内容后面的教程继续更新

不懂的内容问我就可以，加我备注入群可以加入量化交流群，我专业技术支持

量化开户服务需要的找我就可以