等级: 免费版
- 注册:
- 2023-9-29
- 曾用名:
|
我用'AU30'做基准合约, 5分钟级回测时,history_bars 获取的60m级K线时间戳不对.
我的测试代码如下:
#################################################################
# TestLackBars_60m
#################################################################
from PythonApi import *
import datetime
import pandas as pd
import numpy as np
#DEBUG_OUT_PATH = 'F:\\ml_quant\\CheckData\\Output'
DEBUG_OUT_PATH = 'D:\\'
STAMP_INDEX = 0
CLOSE_INDEX = 1
class GlobalCls:
def _init_(self):
self.name = 'global var'
g = GlobalCls()
#PS: 确保g.context已经设置
def Print(content):
dt_str = g.context.now.strftime('%Y%m%d %H:%M:%S')
print("[{}] {}".format(dt_str, content))
#无K线时标的文件日志打印
def FPrint(content):
dt_str = g.context.now.strftime('%Y%m%d %H:%M:%S')
print("[{}] {}".format(dt_str, content))
#log_debug_info(g.debug_log_file, "{0}".format(content))
with open(g.debug_log_file, 'a') as file:
file.write("[{}] {}\n".format(dt_str, content))
#----------------------------------------
pd.set_option('display.max_columns', None) #解决显示列省略问题
pd.set_option('display.max_rows', None)
pd.set_option('display.width', 1000) # 控制，使不换行
pd.set_option('display.float_format', lambda x: '%.2f' % x) #取消科学计数显示
# 在这个方法中编写任何的初始化逻辑。context对象将会在你的算法策略的任何方法之间做传递。--(必须实现)
def init(context):
# 在context中保存全局变量
current_dt_str = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
#日志文件
g.debug_log_file = DEBUG_OUT_PATH + f'\\CheckData_60m_{context.run_info.frequency}_debug_{current_dt_str}.txt'
print("------初始化逻辑init(context)-------")
g.context = context
g.run_type = context.run_info.run_type
print(f"context.run_info.base_book_id:{context.run_info.base_book_id}")
print(f"context.run_info.frequency:{context.run_info.frequency}")
# before_trading此函数会在每天基准合约的策略交易开始前被调用，当天只会被调用一次。--（选择实现）
def before_trading(context):
pass
# 你选择的品种的数据更新将会触发此段逻辑，例如日或分钟历史数据切片或者是实时数据切片更新。--(必须实现)
def handle_bar(context):
current_dt = context.now
symbol_lst = ['LC00', 'SI00', 'AG00']
for symbol in symbol_lst:
freq = '5m'
target_n = 2
bars = history_bars(symbol,target_n, freq, ['datetime', 'close'], include_now=True)
#Print(f"history_bars({symbol}, {target_n}, {freq}, ['datetime', 'close'], include_now=True)")
if len(bars):
bar_stamp = PyramidBarStampToDatetime(bars[-1,STAMP_INDEX])
if bar_stamp != current_dt:
#FPrint(f"Erro: {symbol} {freq} latest bar {bar_stamp} != current_dt:{current_dt}")
continue
#Print(f"\t\t ret bars len:{len(bars)} beg_stamp:{bars[0,0]} end_stamp:{bars[-1,0]}")
freq = '60m'
target_n = 2
bars = history_bars(symbol,target_n, freq, ['datetime', 'close'], include_now=True)
#Print(f"history_bars({symbol}, {target_n}, {freq}, ['datetime', 'close'], include_now=True)")
#if len(bars):
# Print(f"\t\t ret bars len:{len(bars)} beg_stamp:{bars[0,0]} end_stamp:{bars[-1,0]}")
#else:
# Print(f"\t\t ret bars: {bars}")
if len(bars):
#金字塔小时历史数据存在多了yyyymmdd143000数据的bug
if int(bars[-1,STAMP_INDEX]) % 1000000 == 143000: #and (period_data.period == '1h' or period_data.period == '60m'):
FPrint(f"Erro: {symbol} {freq} latest bar {bars[-1,STAMP_INDEX]} ")
continue
# after_trading函数会在每天交易结束后被调用，当天只会被调用一次。 --（选择实现）
def after_trading(context):
pass
def PyramidBarStampToDatetime(val):
val = int(val)
#print("PyramidBarStampToDatetime val:{}".format(val))
return datetime.datetime(year=int(val//10000000000),
month= val%10000000000//100000000,
day= val%100000000//1000000,
hour= val%1000000//10000,
minute=val%10000//100,
second=val%100,
microsecond=0)
|
-
|
发表于 2025-4-21 17:05
发表于 2025-4-21 17:27
楼主