Lean ConnectQuant针对A股的量化数据库搭建【量化数据篇】 ...

本文基于 SQLite 轻量级数据库与 QuantConnect Lean 框架，构建了一套​​支持高频数据回测、策略无缝实盘部署​​的本地化办理方案。
0x01获取所有股票

1. import baostock as bs
2. import pandas as pd
3. import sqlite3
4. from tqdm import tqdm
5. import datetime
7. def save_to_sqlite(df, db_path, table_name):
8.     """将DataFrame存储到SQLite数据库"""
9.     try:
10.         conn = sqlite3.connect(db_path)
11.         df.to_sql(table_name, conn, if_exists='replace', index=False)
12.         print(f"数据成功保存至 {db_path} 的 {table_name} 表")
13.     except Exception as e:
14.         print(f"数据库操作失败：{str(e)}")
15.     finally:
16.         if 'conn' in locals() and conn:
17.             conn.close()
19. def validate_stock_code(code):
20.     """规范股票代码格式为交易所.6位数字（增强创业板/科创板处理）[7,12](@ref)"""
21.     if not isinstance(code, str):
22.         code = str(code)
23.    
24.     code = code.strip().lower().replace(" ", "")
25.    
26.     if code.startswith(("sh.", "sz.", "bj.")):
27.         prefix, num = code.split(".")
28.         return f"{prefix}.{num.zfill(6)}"
29.    
30.     if code.isdigit():
31.         code = code.zfill(6)
32.         first_char = code[0]
33.         if first_char in ['5','6','9','7'] or code.startswith("688"):
34.             return f"sh.{code}"
35.         elif first_char in ['0','1','2','3'] or code.startswith("30"):
36.             return f"sz.{code}"
37.         elif code.startswith("8"):
38.             return f"bj.{code}"
39.    
40.     raise ValueError(f"无法识别的股票代码格式: {code}")
42. def get_index_components(index_func):
43.     """通用指数成分股获取函数（支持自动重试）[8,11](@ref)"""
44.     max_retries = 3
45.     for attempt in range(max_retries):
46.         try:
47.             rs = index_func()
48.             if rs.error_code == '0':
49.                 return {validate_stock_code(item[1]) for item in rs.data}
50.         except Exception as e:
51.             print(f"指数查询第{attempt+1}次失败: {str(e)}")
52.             if attempt == max_retries - 1:
53.                 return set()
54.     return set()
56. def get_all_stocks_with_industry():
57.     """获取全市场股票及指数成分标记"""
58.     try:
59.         # 登录系统
60.         if bs.login().error_code != '0':
61.             raise ConnectionError("Baostock登录失败")
63.         # 获取三大指数成分股（提前获取避免多次查询）[7,11](@ref)
64.         sz50_set = get_index_components(bs.query_sz50_stocks)
65.         hs300_set = get_index_components(bs.query_hs300_stocks)
66.         zz500_set = get_index_components(bs.query_zz500_stocks)
67.         print(f"指数成分股数量：SZ50={len(sz50_set)}, HS300={len(hs300_set)}, ZZ500={len(zz500_set)}")
69.         # 获取全市场股票（含最新交易状态）[13](@ref)
70.         query_date = (datetime.date.today() - datetime.timedelta(days=1)).strftime("%Y-%m-%d")
71.         rs = bs.query_all_stock(day=query_date)
72.         if rs.error_code != '0':
73.             raise ValueError(f"股票查询失败：{rs.error_msg}")
75.         # 处理原始数据（增强状态过滤）[13](@ref)
76.         raw_df = pd.DataFrame(rs.data, columns=rs.fields)
77.         valid_df = raw_df[raw_df['tradeStatus'] == '1']  # 过滤有效交易股票
78.         print(f"初始数据量：{len(raw_df)}，有效交易股票：{len(valid_df)}")
80.         # 代码格式转换
81.         valid_df['valid_code'] = valid_df['code'].apply(lambda x: validate_stock_code(x))
82.         
83.         # 获取行业数据（批量查询优化）[3,9](@ref)
84.         industry_rs = bs.query_stock_industry()
85.         industry_data = []
86.         fields = industry_rs.fields
87.         while industry_rs.next():
88.             row = industry_rs.get_row_data()
89.             try:
90.                 industry_data.append({
91.                     'valid_code': validate_stock_code(row[1]),
92.                     'industry': row[3],
93.                     'industry_type': row[4]
94.                 })
95.             except Exception as e:
96.                 print(f"跳过无效行业数据: {row[1]}, 原因: {str(e)}")
97.         industry_df = pd.DataFrame(industry_data)
99.         # 合并数据
100.         merged_df = pd.merge(
101.             valid_df[['valid_code', 'code_name']].rename(columns={'valid_code':'code', 'code_name':'name'}),
102.             industry_df,
103.             left_on='code',
104.             right_on='valid_code',
105.             how='left'
106.         ).drop(columns=['valid_code'])
108.         # 标记指数成分[7,11](@ref)
109.         merged_df['issz50'] = merged_df['code'].isin(sz50_set).astype(int)
110.         merged_df['ishs300'] = merged_df['code'].isin(hs300_set).astype(int)
111.         merged_df['iszz500'] = merged_df['code'].isin(zz500_set).astype(int)
113.         # 拆分交易所代码
114.         merged_df['exchange'] = merged_df['code'].str.split('.').str[0]
115.         merged_df['code'] = merged_df['code'].str.split('.').str[1]
116.         
117.         # 填充缺失值
118.         merged_df.fillna({
119.             'industry': '未知',
120.             'industry_type': '未分类'
121.         }, inplace=True)
123.         return merged_df[[
124.             'code', 'name', 'industry', 'industry_type',
125.             'issz50', 'ishs300', 'iszz500', 'exchange'
126.         ]]
128.     except Exception as e:
129.         print(f"处理失败：{str(e)}")
130.         return None
131.     finally:
132.         bs.logout()
135. if __name__ == "__main__":
136.     df = get_all_stocks_with_industry()
137.     if df is not None:
138.         print("\n数据样例:")
139.         print(df[df['industry'] != '未知'].head(5))
140.         save_to_sqlite(df, r"../project/data/AAshares/code.db", "all_stocks")

复制代码

0x02 获取日K线数据

1. import baostock as bs
2. import pandas as pd
3. import sqlite3
4. from tqdm import tqdm
5. import time
6. from datetime import datetime
8. def get_stock_codes(conn):
9.     """从数据库获取带交易所前缀的股票代码（结构不变）"""
10.     df = pd.read_sql('SELECT exchange, code FROM all_stocks', conn)
11.     return [f"{row['exchange'].lower().strip()}.{str(row['code']).zfill(6)}"
12.             for _, row in df.iterrows()]
14. def download_daily_data(code, start_date, end_date, max_retries=3):
15.     """下载日K线数据（仅修改frequency和fields）"""
16.     for attempt in range(max_retries):
17.         rs = bs.query_history_k_data_plus(
18.             code=code,
19.             fields="date,code,open,high,low,close,volume,amount",
20.             start_date=start_date,
21.             end_date=end_date,
22.             frequency="d",  # 改为日线[1,7](@ref)
23.             adjustflag="2"   # 保持前复权
24.         )
25.         if rs.error_code == '0':
26.             return rs
27.         time.sleep(1)
28.     return rs
30. def process_daily_data(rs):
31.     """处理日K线数据（移除时间合并步骤）"""
32.     data_list = []
33.     while rs.next():
34.         data_list.append(rs.get_row_data())
35.     df = pd.DataFrame(data_list, columns=rs.fields)
36.    
37.     # 数据清洗（仅保留日期处理）
38.     df['code'] = df['code'].str.split('.').str[1]
39.     numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'amount']
40.     df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce')
41.     return df.dropna().reset_index(drop=True)
43. def save_to_db(df, conn):
44.     # 将DataFrame转换为字典列表
45.     records = df.to_dict('records')
47.     for record in records:
48.         try:
49.             # 单条插入语句
50.             conn.execute(
51.                 """
52.                 INSERT OR IGNORE INTO stock_day_k
53.                 (code, date, open, close, high, low, volume, amount)
54.                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
55.                 """,
56.                 (record['code'], record['date'],
57.                 record['open'], record['close'],
58.                 record['high'], record['low'],
59.                 record['volume'], record['amount'])
60.             )
61.         except IntegrityError:
62.             print(f"主键冲突已跳过：{record['code']} - {record['date']}")
63.             continue
64.     conn.commit()
66. def main():
67.     # 初始化数据库（修改表结构）[5](@ref)
68.     conn = sqlite3.connect(r"../project/data/AAshares/code.db")
69.     conn.executescript('''
70.         CREATE TABLE IF NOT EXISTS stock_day_k (
71.             date TEXT, code TEXT, open REAL, high REAL,
72.             low REAL, close REAL, volume INTEGER, amount REAL,
73.             PRIMARY KEY (date, code)
74.         );
75.         CREATE INDEX IF NOT EXISTS idx_day ON stock_day_k(code);
76.     ''')
78.     # 登录BaoStock（保持原登录逻辑）
79.     if bs.login().error_code != '0':
80.         print("登录失败")
81.         return
83.     try:
84.         codes = get_stock_codes(conn)
85.         start_date = '2025-01-01'  # 保持时间范围不变
86.         end_date = '2025-04-01'
87.         
88.         for code in tqdm(codes, desc="下载进度"):
89.             try:
90.                 rs = download_daily_data(code, start_date, end_date)
91.                 if rs.error_code != '0': continue
92.                
93.                 df = process_daily_data(rs)
94.                 if not df.empty:
95.                     save_to_db(df, conn)
96.                     
97.                 time.sleep(0.5)  # 保持反爬策略不变[8](@ref)
98.                
99.             except Exception as e:
100.                 print(f"{code} 下载失败: {str(e)}")
101.                
102.     finally:
103.         conn.close()
104.         bs.logout()
106. if __name__ == "__main__":
107.     main()

复制代码

0x03 获取5分钟K线数据

1. import baostock as bs
2. import pandas as pd
3. import sqlite3
4. from tqdm import tqdm
5. import time
6. from datetime import datetime
8. def get_stock_codes(conn):
9.     """从数据库获取带交易所前缀的股票代码（网页6最佳实践）"""
10.     df = pd.read_sql('SELECT exchange, code FROM all_stocks', conn)
11.     return [f"{row['exchange'].lower().strip()}.{str(row['code']).zfill(6)}"
12.             for _, row in df.iterrows()]
14. def download_5min_data(code, start_date, end_date, max_retries=3):
15.     """下载5分钟K线数据（含错误重试机制，网页8优化）"""
16.     for attempt in range(max_retries):
17.         rs = bs.query_history_k_data_plus(
18.             code=code,
19.             fields="date,time,code,open,high,low,close,volume,amount",
20.             start_date=start_date,
21.             end_date=end_date,
22.             frequency="5",
23.             adjustflag="2"  # 前复权
24.         )
25.         if rs.error_code == '0':
26.             return rs
27.         time.sleep(1)
28.     return rs
30. def process_5min_data(rs):
31.     """处理5分钟数据格式（网页7时间处理优化）"""
32.     data_list = []
33.     while rs.next():
34.         data_list.append(rs.get_row_data())
35.     df = pd.DataFrame(data_list, columns=rs.fields)
36.    
37.     # 合并日期和时间字段
38.     df['datetime'] = df['time'].apply(
39.         lambda x: f"{x[:4]}-{x[4:6]}-{x[6:8]} {x[8:10]}:{x[10:12]}:00")
40.     df = df.drop(columns=['date', 'time'])
41.     df.rename(columns={'datetime': 'date'}, inplace=True)
42.     # 数据清洗
43.     df['code'] = df['code'].str.split('.').str[1]
44.     numeric_cols = ['open', 'high', 'low', 'close', 'volume','amount']
45.     df[numeric_cols] = df[numeric_cols].apply(pd.to_numeric, errors='coerce')
46.     return df.dropna().reset_index(drop=True)
48. def save_to_db(df, conn):
49.     # 将DataFrame转换为字典列表
50.     records = df.to_dict('records')
52.     for record in records:
53.         try:
54.             # 单条插入语句
55.             conn.execute(
56.                 """
57.                 INSERT OR IGNORE INTO stock_5min_k
58.                 (code, date, open, close, high, low, volume, amount)
59.                 VALUES (?, ?, ?, ?, ?, ?, ?, ?)
60.                 """,
61.                 (record['code'], record['date'],
62.                 record['open'], record['close'],
63.                 record['high'], record['low'],
64.                 record['volume'], record['amount'])
65.             )
66.         except IntegrityError:
67.             print(f"主键冲突已跳过：{record['code']} - {record['date']}")
68.             continue
69.     conn.commit()
71. def main():
72.     # 初始化数据库（网页6表结构）
73.     conn = sqlite3.connect(r"../project/data/AAshares/code.db")
74.     conn.executescript('''
75.         CREATE TABLE IF NOT EXISTS stock_5min_k (
76.             date TEXT, code TEXT, open REAL, high REAL,
77.             low REAL, close REAL, volume INTEGER,amount REAL,
78.             PRIMARY KEY (date, code)
79.         );
80.         CREATE INDEX IF NOT EXISTS idx_5min ON stock_5min_k(code);
81.     ''')
83.     # 登录BaoStock（网页7规范）
84.     if bs.login().error_code != '0':
85.         print("登录失败")
86.         return
88.     try:
89.         codes = get_stock_codes(conn)
90.         start_date = '2025-04-19'
91.         end_date = datetime.now().strftime('%Y-%m-%d')  # 格式为 'YYYY-MM-DD'
92.         
93.         for code in tqdm(codes, desc="下载进度"):
94.             try:
95.                 rs = download_5min_data(code, start_date, end_date)
96.                 if rs.error_code != '0': continue
97.                
98.                 df = process_5min_data(rs)
99.                 if not df.empty:
100.                     save_to_db(df, conn)
101.                     
102.                 time.sleep(0.5)  # 反爬策略（网页8建议）
103.                
104.             except Exception as e:
105.                 print(f"{code} 下载失败: {str(e)}")
106.                
107.     finally:
108.         conn.close()
109.         bs.logout()
111. if __name__ == "__main__":
112.     main()

复制代码

0x04 通过web将获取的数据读出来

1. import csv
2. from io import StringIO
3. from flask import Flask, jsonify, request, Response
4. import sqlite3
5. from datetime import datetime, timedelta, timezone
6. import pandas as pd
7. import requests
9. app = Flask(__name__)
11. def get_db_connection():
12.     conn = sqlite3.connect(r"../project/data/AAshares/code.db")
13.     conn.row_factory = sqlite3.Row
14.     return conn
16. def getdatasqlite(code,table_name):
17.     """统一处理数据请求"""
18.     conn = get_db_connection()
19.     cursor = conn.cursor()
20.     cursor.execute(f"""
21.         SELECT
22.             date AS Date,         
23.             open AS Open,
24.             close AS Close,
25.             high AS High,
26.             low AS Low,
27.             volume AS Volume,
28.             amount AS Amount
29.         FROM {table_name}
30.         WHERE code = ?
31.         ORDER BY date
32.     """, (code,))
33.    
34.     data = cursor.fetchall()
35.     converted_data = []
36.     for row in data:
37.         # 按新的字段顺序解析
38.         date_str = row[0]  # 直接使用数据库返回的日期字符串[3,8](@ref)
39.         # 添加时区转换逻辑（假设数据库存储UTC时间）
40.         if(table_name=="stock_5min_k"):
41.             local_time = datetime.strptime(date_str, "%Y-%m-%d %H:%M:%S")  # 添加秒并解析为本地时间
42.         else:
43.             local_time = datetime.strptime(date_str, "%Y-%m-%d")  # 添加秒并解析为本地时间      
44.         date_time_str = local_time.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")  # 转换为UTC时间
46.         converted = {
47.             "Date": date_time_str,  # 统一时间格式[5,7](@ref)
48.             "Open": float(row[1]),
49.             "Close": float(row[2]),
50.             "High": float(row[3]),
51.             "Low": float(row[4]),
52.             "Volume": float(row[5]),  # 交易量
53.             "Amount": float(row[6])   # 修正字段索引[2,6](@ref)
54.         }
55.         converted_data.append(converted)
56.     return converted_data
59. def getdatahttp(code,table_name):
60.     # 请求第三方接口
61.     if code.startswith("6"):
62.         secid = "1." + code
63.     else:
64.         secid = "0." + code
65.     # end_date = (datetime.now() - timedelta(days=1)).strftime('%Y%m%d')
66.     end_date = (datetime.now()).strftime('%Y%m%d')
67.     if(table_name=="stock_5min_k"):
68.         api_url = f"https://push2his.eastmoney.com/api/qt/stock/kline/get?secid={secid}&fields1=f1%2Cf2%2Cf3%2Cf4%2Cf5%2Cf6&fields2=f51%2Cf52%2Cf53%2Cf54%2Cf55%2Cf56%2Cf57%2Cf58%2Cf59%2Cf60%2Cf61&klt=5&fqt=1&end={end_date}&lmt=1488"  # 替换为实际API地址
69.     else:
70.         api_url = f"https://push2his.eastmoney.com/api/qt/stock/kline/get?secid={secid}&fields1=f1%2Cf2%2Cf3%2Cf4%2Cf5%2Cf6&fields2=f51%2Cf52%2Cf53%2Cf54%2Cf55%2Cf56%2Cf57%2Cf58%2Cf59%2Cf60%2Cf61&klt=101&fqt=1&end={end_date}&lmt=1488"  # 替换为实际API地址
72.     try:
73.         response = requests.get(api_url, timeout=10)
74.         response.raise_for_status()
75.     except requests.exceptions.RequestException as e:
76.         return {"error": f"API请求失败: {str(e)}"}, 500
78.     # 解析原始数据
79.     raw_data = response.json()
80.     klines = raw_data.get('data', {}).get('klines', [])
81.    
82.     # 数据清洗转换
83.     converted_data = []
84.     for item in klines:
85.         parts = item.split(',')
86.         if len(parts) < 11:
87.             continue  # 跳过无效数据
88.             
89.         # 按字段位置解析数据
90.         # 按字段位置解析数据
91.         if(table_name=="stock_5min_k"):
92.             local_time = datetime.strptime(parts[0] + ":00", "%Y-%m-%d %H:%M:%S")  # 添加秒并解析为本地时间
93.         else:
94.             local_time = datetime.strptime(parts[0] , "%Y-%m-%d")  # 添加秒并解析为本地时间
95.         date_time_str = local_time.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")  # 转换为UTC时间
96.         converted = {
97.             "Date": date_time_str,
98.             "Open": float(parts[1]),
99.             "Close": float(parts[2]),
100.             "High": float(parts[3]),
101.             "Low": float(parts[4]),
102.             "Volume": float(parts[5]),  # 交易量
103.             "Amount": float(parts[6]),    # 交易额
104.         }
105.         converted_data.append(converted)
106.     return converted_data
108. def handle_request(code, table_name, format_type='json'):
109.     converted_data = getdatasqlite(code,table_name)
110.     # ...后续格式转换逻辑保持不变...
111.     # 返回数据
112.     if not converted_data:
113.         return {"error": "没有有效数据"}, 404
115.     # 根据要求格式化输出
116.     if format_type == 'csv':
117.         # 生成CSV
118.         si = StringIO()
119.         writer = csv.DictWriter(si, fieldnames=converted_data[0].keys())
120.         writer.writeheader()
121.         writer.writerows(converted_data)  # 写入多行数据
122.         return si.getvalue(), 200, {'Content-Type': 'text/csv'}
123.         
124.     return jsonify(converted_data[-1]), 200
126. @app.route('/dayapi', methods=['GET'])
127. def get_daystock_data():
128.     code = request.args.get('code')
129.     format_type = request.args.get('format', default='json')
130.     if not code:
131.         return jsonify({"error": "缺少参数: code"}), 400
132.     return handle_request(code, 'stock_day_k', format_type)
134. @app.route('/api', methods=['GET'])
135. def get_stock_data():
136.     code = request.args.get('code')
137.     format_type = request.args.get('format', default='json')
138.     if not code:
139.         return jsonify({"error": "缺少参数: code"}), 400
140.     return handle_request(code, 'stock_5min_k', format_type)
142. if __name__ == '__main__':
143.     app.run(host='0.0.0.0', port=80, debug=True)

复制代码

访问 http://ip/dayapi?code={Value} 获取最新的一根K线
访问 http://ip/dayapi?code={Value}&format=csv 返回的是汗青的K线csv
如果成功了，则完成了基础数据的搭建，这个web是为了给Lean QuantConnect调用的。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。

继续阅读请点击广告