数据处理的流程：
1.对数据进行复权（只保留 2016-01-01（含）之后的数据）。
2.验证复权的正确性。
3.由于vnpy的symbol格式是RB999.SHFE这样的格式，所以建立一个期货代码与交易所关系的symbol.json
4.写入SQLite数据库。
最后只要复制 write_to_db.py、symbol.json这2个文件到期货1分钟数据的目录下，运行，它就可以将数据自动复权、并写入到SQLite数据库。
一、对没有复权的主连合约进行复权
我用的是加减复权，代码（并写入数据库write_to_db.py）
import pandas as pd
import sqlite3 # 新增导入
from pathlib import Path
def get_file_prefixes(directory=".", recursive=False, prefix_mode="stem", sep=None, length=None, extensions=None, return_path=False):
"""
遍历目录下的所有文件，提取文件名前缀。
参数:
directory: 目标目录，默认为当前目录 (".")
recursive: 是否递归子目录，默认 False（仅当前目录）
prefix_mode: 前缀提取模式
- "stem": 返回不带扩展名的文件名（如 "file.txt" -> "file"）
- "split": 按分隔符 sep 分割，取第一部分（如 "data_2026.csv" -> "data"）
- "length": 取文件名的前 length 个字符（不包含扩展名，若 length 指定）
sep: 当 mode="split" 时使用的分隔符，默认为 "_"
length: 当 mode="length" 时截取的长度
extensions: 指定要处理的文件扩展名元组，如 ('.csv', '.txt')，默认为 ('.csv',)
return_path: 是否返回完整路径，默认为 False（仅返回前缀字符串列表）
返回:
若 return_path=False: 返回前缀字符串列表（如 ['PT', 'data']）
若 return_path=True: 返回 (文件路径, 前缀) 元组列表
"""
if extensions is None:
extensions = ('.csv',)
extensions = tuple(ext.lower() for ext in extensions)
base = Path(directory)
if not base.exists() or not base.is_dir():
raise ValueError(f"目录不存在或不是文件夹: {directory}")
file_iter = base.rglob("*") if recursive else base.glob("*")
result = []
for item in file_iter:
if item.is_file() and item.suffix.lower() in extensions:
name = item.name
if prefix_mode == "stem":
prefix = item.stem
elif prefix_mode == "split":
split_char = sep if sep is not None else "_"
prefix = name.split(split_char)[0] if split_char in name else name
elif prefix_mode == "length":
stem = item.stem
prefix = stem[:length] if length is not None else stem
else:
raise ValueError("prefix_mode 必须是 'stem', 'split' 或 'length'")
if return_path:
result.append((str(item), prefix))
else:
result.append(prefix)
return result[0]
import json
def find_exchange_from_json(symbol, json_file='symbol.json'):
with open(json_file, 'r', encoding='utf-8') as f:
mapping = json.load(f)
for exchange, symbols in mapping.items():
if symbol in symbols:
return exchange
return None
# ---------- 1. 读取 ----------
file_prefix = get_file_prefixes
df = pd.read_csv(file_prefix+'.csv', parse_dates=['datetime'])
# 只保留 2016-01-01（含）之后的数据
df = df[df['datetime'] >= '2016-01-01']
df.sort_values('datetime', inplace=True)
df.reset_index(drop=True, inplace=True)
# ---------- 2. 保存原始收盘价 ----------
df['close_raw'] = df['close']
# ---------- 3. 提取关键列为列表（脱离DataFrame） ----------
symbols = df['symbol'].tolist()
opens = df['open'].tolist()
highs = df['high'].tolist()
lows = df['low'].tolist()
closes = df['close'].tolist()
close_raws = df['close_raw'].tolist()
# ---------- 4. 等差后复权 ----------
cum_adj = 0.0
last_symbol = None
last_close_raw = None
new_opens = []
new_highs = []
new_lows = []
new_closes = []
for i in range(len(df)):
sym = symbols[i]
if last_symbol is not None and sym != last_symbol:
# 换月跳空差值 = 旧合约最后收盘 - 新合约第一开盘
delta = last_close_raw - opens[i]
cum_adj += delta
# 应用累积因子
new_opens.append(opens[i] + cum_adj)
new_highs.append(highs[i] + cum_adj)
new_lows.append(lows[i] + cum_adj)
new_closes.append(closes[i] + cum_adj)
# 更新状态（用的是原始收盘价，即 close_raws[i]）
last_symbol = sym
last_close_raw = close_raws[i]
# ---------- 5. 覆盖原列 ----------
df['open'] = new_opens
df['high'] = new_highs
df['low'] = new_lows
df['close'] = new_closes
# ---------- 6. 保存 ----------
#df.to_csv('rb_result.csv', index=False)
#print("复权完成，结果已保存至 rb_result.csv")
# ---------- 6. 保存 ----------
# 获取期货商品编码（前缀）
symbol = file_prefix + '999' # 拼接“999”作为 symbol
# 查找交易所
exchange = find_exchange_from_json(file_prefix)
# 构建待入库的 DataFrame，字段名与数据库表完全对应
df_db = pd.DataFrame({
'symbol': symbol, # 统一为 RB999
'exchange': exchange, # 从 json 映射获得
'datetime': df['datetime'], # 直接复制
'interval': '1m', # 固定为 1m
'volume': df['volume'],
'turnover': df['amount'],
'open_interest': df['position'],
'open_price': df['open'],
'high_price': df['high'],
'low_price': df['low'],
'close_price': df['close']
})
# 连接数据库（路径固定）
db_path = r'C:\Users\Administrator\.vntrader\database.db'
conn = sqlite3.connect(db_path)
# 追加写入（表不存在时自动创建）
df_db.to_sql('dbbardata', conn, if_exists='append', index=False)
conn.close()
print("复权完成，数据已保存至 database.db 的 dbbardata 表")
验证1：
在3.11 21:00的时候，开盘价是2000，将它复权成前一分钟的收盘价2085，所以都要加上85，复权后的数据变成这样。
验证2：
在8/17/2016 21:00的时候，开盘价是2531，将它复权成前一分钟的收盘价2589，要加上58.
核心逻辑：没有问题 ✅
-
换月判定（sym != last_symbol）✅
delta 用原始收盘/开盘计算、cum_adj 单调累加 ✅（数学自洽）
换月后整段平移、volume/amount/position 不动 ✅（只有价格需复权）
样例里 JM1309 只出现 2013-06-28 一段、JM1401 只出现 2013-07-01 起一段——我之前担心的两个隐患都不触发：
✅ 无"合约全历史拼接"（新合约换月前没有历史行需平移）
✅ 无同 symbol 重复段（不会重复调整）
一个需要你决策的权衡点（不是 bug，但影响回测）
任何对齐方式都无法让 open 和 close 同时连续，必须二选一：
开盘对齐（你现在的）：开盘价连续，但换月那分钟的 close-to-close 收益有 +0.2% 假跳空；
收盘对齐：收盘连续，换月第一分钟 open 有假跳空。
你的数据是分钟级，如果策略按分钟收益（close/close-1）或把分钟聚合成日线算日收益，换月那分钟/那天的收益会被 +0.2% 假跳空污染。建议根据你的策略用哪种价格确认对齐方式——不是代码错，是设计要选对。
一个方法固有的警告（长周期尤其要注意）
等差后复权对13 年分钟级数据（2013→2026）会累积巨大的 delta，历史价格水平被严重平移（比如 2013 年的 1000 元可能被调成 1800 元）。分钟/日收益率不受影响（同一合约内加同一常数，差分抵消），但所有依赖绝对价格的逻辑会坏：固定止损价、挂单价、ATR 百分比、涨跌停判断等。如果你只用收益率做因子，可以接受；如果策略里有价格阈值，建议用比例后复权对比验证。
附对csv数据进行指定查询的代码：
import pandas as pd
# ---------- 1. 读取数据 ----------
df = pd.read_csv('RB.csv', parse_dates=['datetime'])
df.sort_values('datetime', inplace=True)
df.set_index('datetime', inplace=True)
# ---------- 2. 查询函数（前后各 5 条） ----------
def get_record_with_neighbors(time_str, n=5):
"""
根据时间字符串返回该行及其前后各 n 行（默认 n=5）。
若时间不在数据中，则提示并返回 None。
"""
try:
target = pd.to_datetime(time_str)
except Exception:
print("输入的时间格式无效，请使用如 '1/4/2016 9:09' 的格式。")
return None
if target not in df.index:
print(f"数据中不存在时间为 {time_str} 的记录。")
return None
idx = df.index.get_loc(target) # 目标在排序后索引中的位置（整数）
start = max(0, idx - n) # 向前 n 条，但不小于 0
end = min(len(df) - 1, idx + n) # 向后 n 条，但不超出总行数
result = df.iloc[start:end+1] # 切片左闭右闭
return result
# ---------- 3. 交互示例 ----------
if __name__ == "__main__":
user_input = input("请输入时间（如 1/4/2016 9:09）：")
output = get_record_with_neighbors(user_input, n=5)
if output is not None:
print(f"\n查询结果（包含前 5 条、目标、后 5 条）：")
print(output)
如何自己进行处权处理？
https://www.zhihu.com/question/276668028/answer/3262165543
https://www.vnpy.com/forum/topic/34756-ctaliang-hua-tong-guan-xi-lie-3-lian-xu-he-yue-mei-gao-dui-hui-ce-jie-lun-kao-bu-zhu
二、期货代码与交易所的关系symbol.json
{
"CFFEX": [
"IC",
"IF",
"IH",
"IM",
"T",
"TF",
"TL",
"TS"
],
"CZCE": [
"AP",
"CF"
],
}
三、提取文件名的前缀
from pathlib import Path
def get_file_prefixes(directory=".", recursive=False, prefix_mode="stem", sep=None, length=None, extensions=None, return_path=False):
"""
遍历目录下的所有文件，提取文件名前缀。
参数:
directory: 目标目录，默认为当前目录 (".")
recursive: 是否递归子目录，默认 False（仅当前目录）
prefix_mode: 前缀提取模式
- "stem": 返回不带扩展名的文件名（如 "file.txt" -> "file"）
- "split": 按分隔符 sep 分割，取第一部分（如 "data_2026.csv" -> "data"）
- "length": 取文件名的前 length 个字符（不包含扩展名，若 length 指定）
sep: 当 mode="split" 时使用的分隔符，默认为 "_"
length: 当 mode="length" 时截取的长度
extensions: 指定要处理的文件扩展名元组，如 ('.csv', '.txt')，默认为 ('.csv',)
return_path: 是否返回完整路径，默认为 False（仅返回前缀字符串列表）
返回:
若 return_path=False: 返回前缀字符串列表（如 ['PT', 'data']）
若 return_path=True: 返回 (文件路径, 前缀) 元组列表
"""
if extensions is None:
extensions = ('.csv',)
extensions = tuple(ext.lower() for ext in extensions)
base = Path(directory)
if not base.exists() or not base.is_dir():
raise ValueError(f"目录不存在或不是文件夹: {directory}")
file_iter = base.rglob("*") if recursive else base.glob("*")
result = []
for item in file_iter:
if item.is_file() and item.suffix.lower() in extensions:
name = item.name
if prefix_mode == "stem":
prefix = item.stem
elif prefix_mode == "split":
split_char = sep if sep is not None else "_"
prefix = name.split(split_char)[0] if split_char in name else name
elif prefix_mode == "length":
stem = item.stem
prefix = stem[:length] if length is not None else stem
else:
raise ValueError("prefix_mode 必须是 'stem', 'split' 或 'length'")
if return_path:
result.append((str(item), prefix))
else:
result.append(prefix)
return result[0]
# 示例1：只取当前目录下的文件，返回去掉扩展名的前缀（最常用）
prefixes = get_file_prefixes(directory=".", recursive=False, prefix_mode="stem")
print(prefixes)
import json
def find_exchange_from_json(symbol, json_file='symbol.json'):
with open(json_file, 'r', encoding='utf-8') as f:
mapping = json.load(f)
for exchange, symbols in mapping.items():
if symbol in symbols:
return exchange
return None
a = find_exchange_from_json(prefixes)
print(a)
三、读取vnpy database.db，并划k线
输入起止日期。
经核对，MA999与我Twenty02Strategy策略中生成的日线一致。
import sqlite3
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
from datetime import datetime
import sys
def fetch_daily_kline_interactive(db_path, symbol, start_date, end_date, table_name='dbbardata'):
"""
从数据库读取1分钟数据，合成日K（夜盘归属次日，周五夜盘归下周一），使用Plotly绘制交互式K线图
"""
conn = sqlite3.connect(db_path)
# 探测时间字段类型
cursor = conn.cursor()
cursor.execute(f"SELECT datetime FROM {table_name} LIMIT 1;")
sample = cursor.fetchone()
if sample is None:
conn.close()
raise ValueError(f"表 {table_name} 中无数据。")
time_sample = sample[0]
is_timestamp = isinstance(time_sample, (int, float))
if is_timestamp:
time_condition = "date(datetime, 'unixepoch') BETWEEN ? AND ?"
else:
time_condition = "date(datetime) BETWEEN ? AND ?"
query = f"""
SELECT *
FROM {table_name}
WHERE symbol = ? AND interval = '1m'
AND {time_condition}
ORDER BY datetime ASC
"""
df = pd.read_sql_query(query, conn, params=(symbol, start_date, end_date))
conn.close()
if df.empty:
print(f"在 {start_date} ~ {end_date} 期间未找到 {symbol} 的1分钟数据。")
return
# 时间处理
if is_timestamp:
df['datetime'] = pd.to_datetime(df['datetime'], unit='s')
else:
df['datetime'] = pd.to_datetime(df['datetime'])
df.set_index('datetime', inplace=True)
# ====== 自定义交易日归属（夜盘归属次日，周五夜盘归属下周一）=======
def get_trade_date(dt):
date_ = dt.date()
if dt.hour < 21:
# 白天时段：归属当天
return date_
else:
# 夜盘时段：判断是否为周五
if date_.weekday() == 4: # 周五
return date_ + pd.Timedelta(days=3) # 下周一
else:
return date_ + pd.Timedelta(days=1) # 次日
df['trade_date'] = df.index.map(get_trade_date)
# 按交易日聚合
ohlc_dict = {
'open_price': 'first',
'high_price': 'max',
'low_price': 'min',
'close_price': 'last',
'volume': 'sum'
}
exist_cols = [col for col in ohlc_dict.keys() if col in df.columns]
ohlc_dict = {col: ohlc_dict[col] for col in exist_cols}
daily = df.groupby('trade_date').agg(ohlc_dict).dropna()
daily.reset_index(inplace=True) # trade_date 变为列
daily.rename(columns={'trade_date': 'datetime'}, inplace=True) # 绘图时使用
# 构建交互式图表（主图K线 + 副图成交量）
fig = make_subplots(
rows=2, cols=1,
shared_xaxes=True,
vertical_spacing=0.03,
row_heights=[0.7, 0.3]
)
# 主图：K线
fig.add_trace(
go.Candlestick(
x=daily['datetime'],
open=daily['open_price'],
high=daily['high_price'],
low=daily['low_price'],
close=daily['close_price'],
name='日K线',
showlegend=True
),
row=1, col=1
)
# 副图：成交量（涨红跌绿）
colors = ['red' if daily['close_price'].iloc[i] < daily['open_price'].iloc[i]
else 'green' for i in range(len(daily))]
fig.add_trace(
go.Bar(
x=daily['datetime'],
y=daily['volume'],
name='成交量',
marker_color=colors,
showlegend=True
),
row=2, col=1
)
# 布局设置
fig.update_layout(
title=f'{symbol} 日K线（夜盘归属次日，周五夜盘归下周一） ({start_date} ~ {end_date})',
xaxis_rangeslider_visible=False,
template='plotly_dark',
height=700,
hovermode='x unified'
)
config = {
'scrollZoom': True,
'displayModeBar': True,
'modeBarButtonsToRemove': ['toImage']
}
fig.show(config=config)
return daily
if __name__ == "__main__":
try:
start = input("请输入开始日期 (YYYY-MM-DD): ").strip()
end = input("请输入结束日期 (YYYY-MM-DD): ").strip()
datetime.strptime(start, "%Y-%m-%d")
datetime.strptime(end, "%Y-%m-%d")
except ValueError:
print("日期格式错误，请使用 YYYY-MM-DD 格式。")
sys.exit(1)
fetch_daily_kline_interactive('database.db', 'MA999', start, end)
四、deepseek harness用1分钟k合成日k
合成的数据与我自己写的合成的完全一致。
