Files
quanxiel/quantitative_data/数据批量导入.ipynb
T

742 lines
21 KiB
Plaintext
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
{
"cells": [
{
"cell_type": "markdown",
"metadata": {},
"source": [
"# 量化投资数据批量导入\n",
"\n",
"## 目标\n",
"将 Tushare 的日线行情数据及公司基本面数据批量导入 Docker PostgreSQL (192.168.27.11:5438)\n",
"\n",
"## 数据库结构概览\n",
"\n",
"| 表名 | 说明 | Tushare 接口 |\n",
"|------|------|-------------|\n",
"| stock_basic | 股票基本信息 | stock_basic |\n",
"| trade_cal | 交易日历 | trade_cal |\n",
"| daily | 日线行情 | daily |\n",
"| daily_basic | 每日指标(估值/基本面) | daily_basic |\n",
"| adj_factor | 复权因子 | adj_factor |\n",
"| income | 利润表 | income |\n",
"| balancesheet | 资产负债表 | balancesheet |\n",
"| cashflow | 现金流量表 | cashflow |\n",
"| fina_indicator | 财务指标 | fina_indicator |\n",
"| moneyflow | 个股资金流向 | moneyflow |\n",
"| index_daily | 指数日线行情 | index_daily |\n",
"\n",
"## 使用步骤\n",
"1. 复制 `.env.example` 为 `.env`,填入真实密码和 Token\n",
"2. 逐 Cell 运行本 Notebook(敏感信息通过环境变量读取)"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 0: 检查环境 & 连接测试"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {},
"outputs": [],
"source": [
"import sys\n",
"import os\n",
"os.chdir(r\"d:\\projects\\quanxiel\\quantitative_data\")\n",
"print(f\"工作目录: {os.getcwd()}\")\n",
"print(f\"Python 版本: {sys.version}\")"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {},
"outputs": [],
"source": [
"# 检查依赖包\n",
"!pip list | findstr -i \"tushare pandas psycopg2-binary sqlalchemy python-dotenv\""
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 如果需要安装依赖,取消注释下面这行\n",
"#!pip install -r requirements.txt"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {},
"outputs": [],
"source": [
"# 导入核心模块\n",
"from importer import (\n",
" get_pg_connection,\n",
" get_ts_pro,\n",
" init_database,\n",
" import_stock_basic,\n",
" import_trade_cal,\n",
" import_daily_for_stock,\n",
" import_daily_by_date,\n",
" import_daily_by_year,\n",
" import_daily_basic,\n",
" import_daily_basic_by_date,\n",
" import_adj_factor,\n",
" import_adj_factor_batch,\n",
" import_financial_statements,\n",
" import_index_daily,\n",
" get_all_stock_codes,\n",
" get_stock_codes_from_db,\n",
" full_import,\n",
" batch_insert,\n",
" check_daily_progress,\n",
" check_table_summary,\n",
" get_missing_daily_dates,\n",
" resume_daily_by_date,\n",
" logger,\n",
")\n",
"from config import DB_CONFIG, TUSHARE_TOKEN, START_DATE, END_DATE\n",
"\n",
"print(\"模块导入成功!\")"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {},
"outputs": [],
"source": [
"# 测试数据库连接\n",
"try:\n",
" conn = get_pg_connection()\n",
" cursor = conn.cursor()\n",
" cursor.execute(\"SELECT version()\")\n",
" version = cursor.fetchone()[0]\n",
" print(f\"✓ PostgreSQL 连接成功!\")\n",
" print(f\" 服务器版本: {version}\")\n",
" print(f\" 连接信息: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}\")\n",
" cursor.close()\n",
" conn.close()\n",
"except Exception as e:\n",
" print(f\"✗ 连接失败: {e}\")\n",
" print(\"请检查:\")\n",
" print(\" 1. Docker 容器是否已启动: docker ps | findstr postgres\")\n",
" print(\" 2. 环境变量 (.env) 中的连接参数是否正确\")\n",
" print(\" 3. 防火墙是否开放 5438 端口\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 测试 Tushare API 连接\n",
"try:\n",
" pro = get_ts_pro()\n",
" # 简单测试: 获取一只股票信息\n",
" df = pro.stock_basic(ts_code=\"000001.SZ\", fields=\"ts_code,name,industry\")\n",
" print(f\"✓ Tushare API 连接成功!\")\n",
" print(f\" 测试查询: {df.iloc[0].to_dict()}\")\n",
"except Exception as e:\n",
" print(f\"✗ Tushare API 连接失败: {e}\")\n",
" print(\"请检查环境变量 TUSHARE_TOKEN 是否正确(或 .env 文件)\")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 1: 初始化数据库 Schema"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 执行 DDL,创建所有表结构\n",
"init_database()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 2: 导入股票基本信息"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入全量 A 股股票基本信息(含上市和退市)\n",
"import_stock_basic()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 验证:查看导入结果\n",
"conn = get_pg_connection()\n",
"cursor = conn.cursor()\n",
"cursor.execute(\"SELECT COUNT(*) FROM stock_basic\")\n",
"print(f\"stock_basic 总记录数: {cursor.fetchone()[0]}\")\n",
"cursor.execute(\"SELECT list_status, COUNT(*) FROM stock_basic GROUP BY list_status\")\n",
"for row in cursor.fetchall():\n",
" print(f\" 状态 '{row[0]}': {row[1]} 只\")\n",
"cursor.close()\n",
"conn.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 3: 导入交易日历"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入交易日历 (默认从环境变量 QUANT_START_DATE ~ QUANT_END_DATE)\n",
"import_trade_cal()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 或者指定日期范围\n",
"# import_trade_cal(start_date=\"2020-01-01\", end_date=\"2025-12-31\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 验证\n",
"conn = get_pg_connection()\n",
"cursor = conn.cursor()\n",
"cursor.execute(\"\"\"\n",
" SELECT exchange, MIN(cal_date) AS first_date, MAX(cal_date) AS last_date, COUNT(*) AS total\n",
" FROM trade_cal\n",
" GROUP BY exchange\n",
"\"\"\")\n",
"for row in cursor.fetchall():\n",
" print(f\" {row[0]}: {row[1]} ~ {row[2]}, 共 {row[3]} 条\")\n",
"cursor.close()\n",
"conn.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 4: 导入日线行情 (核心表,最耗时)\n",
"\n",
"> **新版改进:** 按交易日循环拉取全市场数据 `pro.daily_vip(trade_date='20180810')`VIP接口),\n",
"> API 调用从 ~80,000次 (5000只×16年) 降至 ~4,000次 (250交易日×16年),速度提升约 **20倍**。\n",
"\n",
"### 4.1 按年批量导入 (推荐)\n",
"\n",
"数据量估算: ~4000个交易日,每个交易日约5000条记录 ≈ 2000万条记录\n",
"\n",
"> **注意:** 新版 `import_daily_by_year` 不再需要 `stock_list` 参数,内部自动从 `trade_cal` 获取交易日列表后逐日拉取全市场数据。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 按年份逐批导入日线行情 (按交易日循环拉取全市场数据)\n",
"# 不再需要 stock_list 参数 — 自动查询 trade_cal 获取交易日\n",
"# 如果中断,修改年份范围从断点继续即可 (UPSERT 幂等,不会重复)\n",
"import_daily_by_year(\n",
" start_year=2010,\n",
" end_year=2025,\n",
" sleep_interval=0.3, # API 频率控制,免费版建议 0.3~0.5\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 4.1.1 查看日线导入进度\n",
"\n",
"按年份统计 daily 表中已导入的记录数和独立股票数,了解导入到哪个年份了。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 查看每天的日线导入进度(按年份统计)\n",
"check_daily_progress()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 查看所有表的整体概览\n",
"check_table_summary()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 4.1.2 按缺失日期断点续传 (推荐)\n",
"\n",
"新版 `resume_daily_by_date` 自动对比 `trade_cal` 和 `daily` 表,**只导入缺失日期的全市场数据**。\n",
"粒度精确到交易日级别,比旧的按年份续传更精细。\n",
"\n",
"**工作流程:**\n",
"1. 先调用 `get_missing_daily_dates()` 查看缺失的交易日列表\n",
"2. 调用 `resume_daily_by_date()` 仅补缺缺失的交易日"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 第1步:查看缺失的交易日 (仅查询,不导入)\n",
"missing = get_missing_daily_dates(\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
")\n",
"print(f\"缺失交易日总数: {len(missing)}\")\n",
"if len(missing) <= 30:\n",
" print(f\"缺失日期: {missing}\")\n",
"else:\n",
" # 按年份汇总显示\n",
" from collections import Counter\n",
" year_counts = Counter(d[:4] for d in missing)\n",
" for y in sorted(year_counts):\n",
" print(f\" {y} 年: {year_counts[y]} 个缺失交易日\")\n",
" print(f\" (前10个缺失日期): {missing[:10]}\")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 第2步:按缺失日期断点续传,自动补充缺失的交易日数据\n",
"# 例如之前中断了,这里只会导入尚未导入的交易日数据\n",
"resume_daily_by_date(\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
" sleep_interval=0.3,\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"> **备用方案:** 也可以手动指定断点年份重新调用 `import_daily_by_year`,因为 UPSERT 幂等,重复导入已存在的数据不会产生重复记录。\n",
">\n",
"> ```python\n",
"> # 例如假设 2010~2020 已完成,从 2021 年继续\n",
"> import_daily_by_year(start_year=2021, end_year=2025)\n",
"> ```\n",
">\n",
"> 也可查看 `import_data.log` 文件获取最后一次成功的日志输出。"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"### 4.2 单只股票补充导入\n",
"\n",
"如果某只股票数据缺失,可以单独补导(保留原接口兼容)。"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入单只股票日线行情 (用于补充导入或测试)\n",
"conn = get_pg_connection()\n",
"n = import_daily_for_stock(\"000001.SZ\", \"2020-01-01\", \"2020-12-31\", conn)\n",
"print(f\"导入 000001.SZ 2020年数据: {n} 条\")\n",
"conn.close()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 验证日线数据\n",
"conn = get_pg_connection()\n",
"cursor = conn.cursor()\n",
"cursor.execute(\"\"\"\n",
" SELECT \n",
" COUNT(*) AS total_records,\n",
" COUNT(DISTINCT ts_code) AS stock_count,\n",
" MIN(trade_date) AS first_date,\n",
" MAX(trade_date) AS last_date\n",
" FROM daily\n",
"\"\"\")\n",
"for row in cursor.fetchall():\n",
" print(f\" 总记录数: {row[0]:,}\")\n",
" print(f\" 股票数量: {row[1]}\")\n",
" print(f\" 日期范围: {row[2]} ~ {row[3]}\")\n",
"cursor.close()\n",
"conn.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 5: 导入每日指标 (估值数据)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 按交易日导入每日指标 (PE/PB/PS/总市值/流通市值等)\n",
"import_daily_basic_by_date(\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 验证\n",
"conn = get_pg_connection()\n",
"cursor = conn.cursor()\n",
"cursor.execute(\"SELECT COUNT(*), MIN(trade_date), MAX(trade_date) FROM daily_basic\")\n",
"row = cursor.fetchone()\n",
"print(f\" daily_basic: {row[0]:,} 条, {row[1]} ~ {row[2]}\")\n",
"cursor.close()\n",
"conn.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 6: 导入复权因子"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入复权因子 (用于前复权/后复权价格计算)\n",
"# 需要先获取股票列表\n",
"stock_list = get_stock_codes_from_db()\n",
"print(f\"共 {len(stock_list)} 只股票\")\n",
"\n",
"import_adj_factor_batch(\n",
" stock_list,\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 7: 导入财务数据 (三大报表 + 财务指标)\n",
"\n",
"⚠ 此步骤耗时较长,约需数小时(取决于股票数量)"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入利润表、资产负债表、现金流量表、财务指标\n",
"# stock_list 从上一步已获取,或重新获取\n",
"if 'stock_list' not in dir():\n",
" stock_list = get_stock_codes_from_db()\n",
"\n",
"import_financial_statements(\n",
" stock_list,\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
")"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 验证财务数据\n",
"conn = get_pg_connection()\n",
"cursor = conn.cursor()\n",
"for table in [\"income\", \"balancesheet\", \"cashflow\", \"fina_indicator\"]:\n",
" cursor.execute(f\"SELECT COUNT(*) FROM {table}\")\n",
" count = cursor.fetchone()[0]\n",
" print(f\" {table}: {count:,} 条\")\n",
"cursor.close()\n",
"conn.close()"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## Step 8: 导入指数日线行情"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 导入主要指数日线行情\n",
"import_index_daily(\n",
" index_codes=[\n",
" \"000001.SH\", # 上证指数\n",
" \"399001.SZ\", # 深证成指\n",
" \"000300.SH\", # 沪深300\n",
" \"000905.SH\", # 中证500\n",
" \"399006.SZ\", # 创业板指\n",
" \"000688.SH\", # 科创50\n",
" \"000016.SH\", # 上证50\n",
" \"399005.SZ\", # 中小100\n",
" \"000852.SH\", # 中证1000\n",
" ],\n",
" start_date=\"2010-01-01\",\n",
" end_date=\"2025-12-31\",\n",
")"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## 一键全量导入 (可选)\n",
"\n",
"如果不想逐步执行,可以运行下面这个 Cell 一键完成所有导入"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 一键全量导入 (需数小时,请谨慎)\n",
"# full_import(\n",
"# start_date=\"2010-01-01\",\n",
"# end_date=\"2025-12-31\",\n",
"# import_financials=True, # 设为 False 跳过财务数据加快速度\n",
"# )"
]
},
{
"cell_type": "markdown",
"metadata": {},
"source": [
"---\n",
"## 数据验证与查询示例"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"import pandas as pd\n",
"import psycopg2\n",
"\n",
"conn = get_pg_connection()\n",
"\n",
"# 各表统计\n",
"tables = [\"stock_basic\", \"daily\", \"daily_basic\", \"adj_factor\",\n",
" \"income\", \"balancesheet\", \"cashflow\", \"fina_indicator\",\n",
" \"trade_cal\", \"index_daily\"]\n",
"\n",
"print(f\"{'表名':<20} {'记录数':>12} {'最早日期':>12} {'最晚日期':>12}\")\n",
"print(\"-\" * 60)\n",
"for table in tables:\n",
" try:\n",
" count_sql = f\"SELECT COUNT(*) FROM {table}\"\n",
" count = pd.read_sql(count_sql, conn).iloc[0, 0]\n",
" \n",
" # 尝试获取日期范围\n",
" date_col = None\n",
" if table == \"daily\":\n",
" date_col = \"trade_date\"\n",
" elif table == \"daily_basic\":\n",
" date_col = \"trade_date\"\n",
" elif table in [\"income\", \"balancesheet\", \"cashflow\"]:\n",
" date_col = \"end_date\"\n",
" elif table == \"fina_indicator\":\n",
" date_col = \"end_date\"\n",
" elif table == \"trade_cal\":\n",
" date_col = \"cal_date\"\n",
" elif table == \"index_daily\":\n",
" date_col = \"trade_date\"\n",
" elif table == \"adj_factor\":\n",
" date_col = \"trade_date\"\n",
" \n",
" if date_col:\n",
" date_sql = f\"SELECT MIN({date_col}), MAX({date_col}) FROM {table}\"\n",
" min_d, max_d = pd.read_sql(date_sql, conn).iloc[0]\n",
" print(f\"{table:<20} {count:>12,} {str(min_d)[:10]:>12} {str(max_d)[:10]:>12}\")\n",
" else:\n",
" print(f\"{table:<20} {count:>12,}\")\n",
" except Exception as e:\n",
" print(f\"{table:<20} {'错误':>12}: {str(e)[:40]}\")\n",
"\n",
"conn.close()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 示例查询 1: 查询某股票最近10个交易日数据\n",
"query1 = \"\"\"\n",
"SELECT trade_date, open, high, low, close, vol, amount, pct_chg\n",
"FROM daily\n",
"WHERE ts_code = '000001.SZ'\n",
"ORDER BY trade_date DESC\n",
"LIMIT 10\n",
"\"\"\"\n",
"conn = get_pg_connection()\n",
"df1 = pd.read_sql(query1, conn)\n",
"print(\"平安银行(000001.SZ) 最近10个交易日:\")\n",
"display(df1)\n",
"conn.close()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 示例查询 2: 日线行情 + 估值指标联合查询 (使用视图)\n",
"query2 = \"\"\"\n",
"SELECT *\n",
"FROM v_daily_with_valuation\n",
"WHERE ts_code = '000001.SZ'\n",
" AND trade_date >= '2024-01-01'\n",
"ORDER BY trade_date DESC\n",
"LIMIT 10\n",
"\"\"\"\n",
"conn = get_pg_connection()\n",
"df2 = pd.read_sql(query2, conn)\n",
"print(\"平安银行 - 日线+估值:\")\n",
"display(df2[['trade_date', 'close', 'pct_chg', 'pe', 'pe_ttm', 'pb', 'total_mv']])\n",
"conn.close()"
]
},
{
"cell_type": "code",
"execution_count": null,
"metadata": {},
"outputs": [],
"source": [
"# 示例查询 3: 最新财务指标 Top 20 (按 ROE 排序)\n",
"query3 = \"\"\"\n",
"SELECT *\n",
"FROM v_latest_financials\n",
"WHERE roe IS NOT NULL\n",
" AND roe > 0\n",
"ORDER BY roe DESC\n",
"LIMIT 20\n",
"\"\"\"\n",
"conn = get_pg_connection()\n",
"df3 = pd.read_sql(query3, conn)\n",
"print(\"ROE Top 20:\")\n",
"display(df3[['ts_code', 'name', 'industry', 'roe', 'roa', 'eps', 'debt_to_assets']])\n",
"conn.close()"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.10.2"
}
},
"nbformat": 4,
"nbformat_minor": 4
}