一、利用Iqunt
注意：一定要下载复权的数据
（一）运用策略
1.确认并下载 1 分钟历史数据（必须！）
在 iQuant 客户端 操作 → 数据管理 中：
品种：000001.SZ
周期：1分钟
时间范围：覆盖 2024-01-01 至 2024-01-10
点击下载，等待完成。
备注：
电脑上已存2011-2026.7.31的日k线数据。
2.运行策略
导出的文件在D:\iquant_data\
#encoding:gbk
import pandas as pd
import os
def init(ContextInfo):
# ====== 配置参数 ======
ContextInfo.stock_code = '000001.SZ' # 股票代码
ContextInfo.start_date = '20250101' # 开始日期 yyyymmdd
ContextInfo.end_date = '20260110' # 结束日期 yyyymmdd
ContextInfo.output_dir = r'D:\iquant_data' # 输出目录
ContextInfo.export_done = False
def after_init(ContextInfo):
if ContextInfo.export_done:
return
ContextInfo.export_done = True
stock = ContextInfo.stock_code
start = ContextInfo.start_date
end = ContextInfo.end_date
print(f"开始下载 {stock} 从 {start} 到 {end} 的1分钟K线数据...")
stock_name = ContextInfo.get_stock_name(stock)
print(f"股票名称: {stock_name}")
data_dict = ContextInfo.get_market_data_ex(
fields=['open', 'high', 'low', 'close', 'volume', 'amount'],
stock_code=[stock],
period='1m',
start_time=start,
end_time=end,
count=-1,
dividend_type='front',
fill_data=False,
subscribe=False
)
df = data_dict.get(stock)
if df is None or df.empty:
print("未获取到任何数据，请检查日期范围或数据是否已下载。")
return
print(f"共获取 {len(df)} 条数据")
df = df.copy()
df.reset_index(inplace=True)
time_col = 'stime' if 'stime' in df.columns else 'time'
if time_col not in df.columns:
print("错误：未找到时间列")
return
# ---------- 时间转换 ----------
time_str = df[time_col].astype(str)
sample = time_str.iloc[0] if len(time_str) > 0 else ''
if sample.isdigit() and len(sample) == 14:
df['datetime'] = pd.to_datetime(time_str, format='%Y%m%d%H%M%S', errors='coerce')
elif sample.isdigit() and len(sample) == 8:
df['datetime'] = pd.to_datetime(time_str, format='%Y%m%d', errors='coerce')
else:
df[time_col] = pd.to_numeric(df[time_col], errors='coerce')
df = df[df[time_col] > 0]
df['datetime'] = pd.to_datetime(df[time_col], unit='ms', errors='coerce')
df = df.dropna(subset=['datetime'])
print(f"有效数据行数: {len(df)}")
if len(df) == 0:
print("时间转换后无有效数据，请检查数据")
return
# 检查数据实际起止日期
actual_start = df['datetime'].min().strftime('%Y-%m-%d')
actual_end = df['datetime'].max().strftime('%Y-%m-%d')
print(f"实际数据范围: {actual_start} 至 {actual_end}")
if actual_start > start:
print(f"警告：本地数据最早只到 {actual_start}，请补充 {start} 至 {actual_start} 的历史数据")
# ---------- 格式化输出 ----------
# 日期和时间分别格式化为 yyyy-mm-dd 和 HH:MM:SS 字符串
df['日期'] = df['datetime'].dt.strftime('%Y-%m-%d')
df['时间'] = df['datetime'].dt.strftime('%H:%M:%S')
# 价格字段保留2位小数
for col in ['open', 'high', 'low', 'close']:
if col in df.columns:
df[col] = df[col].round(2)
# 成交额一般也保留2位小数（可选）
if 'amount' in df.columns:
df.rename(columns={'amount': '成交额'}, inplace=True)
df['成交额'] = df['成交额'].round(2)
else:
df['成交额'] = 0.0
df['股票代码'] = stock
df['股票名称'] = stock_name
columns_order = ['日期', '时间', '股票代码', '股票名称',
'open', 'high', 'low', 'close', 'volume', '成交额']
existing_cols =
df = df[existing_cols]
# 保存 CSV（注意：strftime 输出的已经是格式化字符串，所以不会自动变回日期对象）
if not os.path.exists(ContextInfo.output_dir):
os.makedirs(ContextInfo.output_dir)
file_name = stock.replace('.', '_') + '.csv'
file_path = os.path.join(ContextInfo.output_dir, file_name)
df.to_csv(file_path, index=False, encoding='gbk')
print(f"数据已保存至: {file_path}")
print("\n数据预览:")
print(df.head())
def handlebar(ContextInfo):
pass
3.核对数据无误
（二）导出日线
# -*- coding: gbk -*-
import pandas as pd
def init(ContextInfo):
# ---------- 1. 获取全部A股列表 ----------
stock_list = ContextInfo.get_stock_list_in_sector('沪深A股') # 若不存在，改为 '全部A股'
# 剔除北交所（代码以4、8开头）
#stock_list = [s for s in stock_list if not (s.split('.')[0].startswith('4') or s.split('.')[0].startswith('8'))]
print(f'总股票数量：{len(stock_list)}')
# ---------- 2. 参数设置 ----------
start_time = '20250101'
end_time = '20260731'
batch_size = 500 # 每批处理股票数
all_dfs = [] # 存放各批次DataFrame
# ---------- 3. 分批获取数据 ----------
for i in range(0, len(stock_list), batch_size):
batch_codes = stock_list[i:i+batch_size]
print(f'正在处理第 {i//batch_size + 1} 批，共 {len(batch_codes)} 只股票...')
try:
data_dict = ContextInfo.get_market_data_ex(
fields=['open', 'high', 'low', 'close', 'volume', 'amount'],
stock_code=batch_codes,
period='1d',
start_time=start_time,
end_time=end_time,
count=-1,
dividend_type='front_ratio',
fill_data=True
)
# 提取有效数据，重置索引以将日期变为列
batch_dfs = []
for code, df in data_dict.items():
if df is not None and not df.empty:
df = df.copy()
df = df.reset_index() # 将日期索引变为普通列
df.rename(columns={'index': 'date'}, inplace=True) # 重命名日期列
df['stock'] = code
batch_dfs.append(df)
if batch_dfs:
batch_df = pd.concat(batch_dfs)
all_dfs.append(batch_df)
print(f'第 {i//batch_size + 1} 批完成，获取 {len(batch_df)} 行')
else:
print(f'第 {i//batch_size + 1} 批无有效数据')
except Exception as e:
print(f'第 {i//batch_size + 1} 批获取失败：{e}')
continue
# ---------- 4. 合并所有批次 ----------
if all_dfs:
final_df = pd.concat(all_dfs, ignore_index=True)
print(f'总数据行数：{len(final_df)}')
# ---------- 5. 保存为 CSV（含日期列） ----------
csv_file = '全A股日线_2025-2026.csv'
final_df.to_csv(csv_file, encoding='gbk', index=False)
print(f'数据已保存为 CSV 文件：{csv_file}')
else:
print('未获取到任何数据，请检查数据下载和板块名称。')
def handlebar(ContextInfo):
pass
1.备注：必须将数据下载到电脑才能使用get_market_data_ex
为什么 很多股票20260819有数据，8.1-8.18都没有？
如果本地没有该日的数据行，get_market_data_ex 只能返回已有的数据行（通常是最新一天）。您看到的情况就是本地缺乏大部分历史日线数据。
2.关于停牌
如果遇到停牌的股票，有ohlc四个价格，但是成交量和金额是空的。这是iquant默认的。
价格填充：为了便于技术指标计算和复权处理，平台在日线数据中对停牌日自动向前填充最近有效价格（即用停牌前最后一个交易日的收盘价作为停牌期间的开高低收）。
成交量/成交额保持为0：停牌日确实没有交易，因此 volume 和 amount 如实记录为0，不做填充。
这是国内绝大多数量化平台（聚宽、优矿、通达信等）的通用做法，并非异常。
（三）csv转parquet
import csv
import pandas as pd
import pyarrow as pa
import pyarrow.parquet as pq
CSV_PATH = "全A股日线_2021-20260731.csv"
PARQUET_PATH = "全A股日线_2021-20260731.parquet"
CHUNK_SIZE = 500_000
ENCODING = "utf-8-sig" # 如果读取报编码错误，可改为 "gbk"
# 自动检测分隔符
with open(CSV_PATH, encoding=ENCODING, newline="") as f:
sample = f.read(4096)
try:
sep = csv.Sniffer().sniff(sample, delimiters=",\t;| ").delimiter
except csv.Error:
sep = ","
if sep == " ":
read_kwargs = {"delim_whitespace": True}
else:
read_kwargs = {"sep": sep}
print(f"检测到分隔符: {sep!r}")
# 使用可空整数类型处理可能缺失的 volume 和 amount
dtype = {
"stime": str,
"open": "float64",
"high": "float64",
"low": "float64",
"close": "float64",
"volume": "Int64",
"amount": "Int64",
"stock": str,
}
reader = pd.read_csv(
CSV_PATH,
encoding=ENCODING,
dtype=dtype,
chunksize=CHUNK_SIZE,
**read_kwargs,
)
writer = None
for chunk in reader:
# stime 转为日期类型；如果想保留原始字符串可删除下一行
chunk["stime"] = pd.to_datetime(chunk["stime"], format="%Y%m%d")
table = pa.Table.from_pandas(chunk, preserve_index=False)
if writer is None:
writer = pq.ParquetWriter(
PARQUET_PATH,
table.schema,
compression="snappy",
)
writer.write_table(table)
if writer is not None:
writer.close()
print(f"转换完成: {PARQUET_PATH}")
（四）检查parquet是否正确
import pandas as pd
PARQUET_PATH = "全A股日线_2021-20260731.parquet"
df = pd.read_parquet(PARQUET_PATH)
# 设置显示选项，避免列被截断
pd.set_option("display.max_columns", None)
pd.set_option("display.width", 200)
pd.set_option("display.max_colwidth", 30)
print("最开始10条记录：")
print(df.head(10))
print("\n最末尾10条记录：")
print(df.tail(10))
（三）存在问题
导出2025-2026数据时，发现有2支股票002605.SZ、002927.SZ的数据是空的，打开iqunat软件，切换到这2支股票，也的确看不到它的日线图。
原来是除权信息惹的祸。
我用单独下载这只股票数据作测试，不除权可以下载，用了dividend_type='front_ratio'也就无法下载。
下载复权数据之后还是不行。
二、通达信导出：
选择标的（螺纹钢主连）-> 选择K线级别（1分钟）-> 导出为txt或excel[reference:0]
按代码：34或者在菜单中选择：
