quanxiel/quantitative_data/.ipynb_checkpoints/数据批量导入-checkpoint.ipynb

量化投资数据批量导入

目标

将 Tushare 的日线行情数据及公司基本面数据批量导入 Docker PostgreSQL (192.168.27.15:12345)

数据库结构概览

表名	说明	Tushare 接口
stock_basic	股票基本信息	stock_basic
trade_cal	交易日历	trade_cal
daily	日线行情	daily
daily_basic	每日指标(估值/基本面)	daily_basic
adj_factor	复权因子	adj_factor
income	利润表	income
balancesheet	资产负债表	balancesheet
cashflow	现金流量表	cashflow
fina_indicator	财务指标	fina_indicator
moneyflow	个股资金流向	moneyflow
index_daily	指数日线行情	index_daily

使用步骤

1. 修改 config.py 中的数据库密码和 Tushare Token
2. 逐 Cell 运行本 Notebook

---

Step 0: 检查环境 & 连接测试

In [ ]:

import sys
import os
os.chdir(r"t:\jupyter\notebook\quantitative_data")
print(f"工作目录: {os.getcwd()}")
print(f"Python 版本: {sys.version}")

In [ ]:

# 检查依赖包
!pip list | findstr -i "tushare pandas psycopg2 sqlalchemy"

In [ ]:

# 如果需要安装依赖，取消注释下面这行
# !pip install -r requirements.txt

In [ ]:

# 导入核心模块
from importer import (
    get_pg_connection,
    get_ts_pro,
    init_database,
    import_stock_basic,
    import_trade_cal,
    import_daily_by_year,
    import_daily_basic,
    import_daily_basic_by_date,
    import_adj_factor,
    import_adj_factor_batch,
    import_financial_statements,
    import_index_daily,
    get_all_stock_codes,
    get_stock_codes_from_db,
    full_import,
    batch_insert,
    logger,
)
from config import DB_CONFIG, TUSHARE_TOKEN, START_DATE, END_DATE

print("模块导入成功!")

In [ ]:

# 测试数据库连接
try:
    conn = get_pg_connection()
    cursor = conn.cursor()
    cursor.execute("SELECT version()")
    version = cursor.fetchone()[0]
    print(f"✓ PostgreSQL 连接成功!")
    print(f"  服务器版本: {version}")
    print(f"  连接信息: {DB_CONFIG['host']}:{DB_CONFIG['port']}/{DB_CONFIG['database']}")
    cursor.close()
    conn.close()
except Exception as e:
    print(f"✗ 连接失败: {e}")
    print("请检查:")
    print("  1. Docker 容器是否已启动: docker ps | findstr postgres")
    print("  2. config.py 中的连接参数是否正确")
    print("  3. 防火墙是否开放 12345 端口")

In [ ]:

# 测试 Tushare API 连接
try:
    pro = get_ts_pro()
    # 简单测试: 获取一只股票信息
    df = pro.stock_basic(ts_code="000001.SZ", fields="ts_code,name,industry")
    print(f"✓ Tushare API 连接成功!")
    print(f"  测试查询: {df.iloc[0].to_dict()}")
except Exception as e:
    print(f"✗ Tushare API 连接失败: {e}")
    print("请检查 config.py 中的 TUSHARE_TOKEN 是否正确")

---

Step 1: 初始化数据库 Schema

In [ ]:

# 执行 DDL，创建所有表结构
init_database()

---

Step 2: 导入股票基本信息

In [ ]:

# 导入全量 A 股股票基本信息（含上市和退市）
import_stock_basic()

In [ ]:

# 验证：查看导入结果
conn = get_pg_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*) FROM stock_basic")
print(f"stock_basic 总记录数: {cursor.fetchone()[0]}")
cursor.execute("SELECT list_status, COUNT(*) FROM stock_basic GROUP BY list_status")
for row in cursor.fetchall():
    print(f"  状态 '{row[0]}': {row[1]} 只")
cursor.close()
conn.close()

---

Step 3: 导入交易日历

In [ ]:

# 导入交易日历 (默认从 config.py 的 START_DATE ~ END_DATE)
import_trade_cal()

In [ ]:

# 或者指定日期范围
# import_trade_cal(start_date="2020-01-01", end_date="2025-12-31")

In [ ]:

# 验证
conn = get_pg_connection()
cursor = conn.cursor()
cursor.execute("""
    SELECT exchange, MIN(cal_date) AS first_date, MAX(cal_date) AS last_date, COUNT(*) AS total
    FROM trade_cal
    GROUP BY exchange
""")
for row in cursor.fetchall():
    print(f"  {row[0]}: {row[1]} ~ {row[2]}, 共 {row[3]} 条")
cursor.close()
conn.close()

---

Step 4: 导入日线行情 (核心表，最耗时)

4.1 获取股票列表

In [ ]:

# 获取所有需要导入的股票代码
stock_list = get_stock_codes_from_db()
print(f"共 {len(stock_list)} 只股票需要导入日线行情")
print(f"前 10 只: {stock_list[:10]}")

4.2 按年批量导入 (推荐 - 断点续传友好)

数据量估算: 约5000只股票 × 250交易日/年 × 16年 ≈ 2000万条记录

In [ ]:

# 按年份逐批导入日线行情
# 如果中断，可以修改年份范围从断点继续
import_daily_by_year(
    stock_list,
    start_year=2010,
    end_year=2025,
)

4.3 单只股票导入 (补充/重试)

In [ ]:

# 导入单只股票日线行情 (用于补充导入或测试)
conn = get_pg_connection()
from importer import import_daily_for_stock

n = import_daily_for_stock("000001.SZ", "2020-01-01", "2020-12-31", conn)
print(f"导入 000001.SZ 2020年数据: {n} 条")
conn.close()

In [ ]:

# 验证日线数据
conn = get_pg_connection()
cursor = conn.cursor()
cursor.execute("""
    SELECT 
        COUNT(*) AS total_records,
        COUNT(DISTINCT ts_code) AS stock_count,
        MIN(trade_date) AS first_date,
        MAX(trade_date) AS last_date
    FROM daily
""")
for row in cursor.fetchall():
    print(f"  总记录数: {row[0]:,}")
    print(f"  股票数量: {row[1]}")
    print(f"  日期范围: {row[2]} ~ {row[3]}")
cursor.close()
conn.close()

---

Step 5: 导入每日指标 (估值数据)

In [ ]:

# 按交易日导入每日指标 (PE/PB/PS/总市值/流通市值等)
import_daily_basic_by_date(
    start_date="2010-01-01",
    end_date="2025-12-31",
)

In [ ]:

# 验证
conn = get_pg_connection()
cursor = conn.cursor()
cursor.execute("SELECT COUNT(*), MIN(trade_date), MAX(trade_date) FROM daily_basic")
row = cursor.fetchone()
print(f"  daily_basic: {row[0]:,} 条, {row[1]} ~ {row[2]}")
cursor.close()
conn.close()

---

Step 6: 导入复权因子

In [ ]:

# 导入复权因子 (用于前复权/后复权价格计算)
import_adj_factor_batch(
    stock_list,
    start_date="2010-01-01",
    end_date="2025-12-31",
)

---

Step 7: 导入财务数据 (三大报表 + 财务指标)

⚠ 此步骤耗时较长，约需数小时（取决于股票数量）

In [ ]:

# 导入利润表、资产负债表、现金流量表、财务指标
import_financial_statements(
    stock_list,
    start_date="2010-01-01",
    end_date="2025-12-31",
)

In [ ]:

# 验证财务数据
conn = get_pg_connection()
cursor = conn.cursor()
for table in ["income", "balancesheet", "cashflow", "fina_indicator"]:
    cursor.execute(f"SELECT COUNT(*) FROM {table}")
    count = cursor.fetchone()[0]
    print(f"  {table}: {count:,} 条")
cursor.close()
conn.close()

---

Step 8: 导入指数日线行情

In [ ]:

# 导入主要指数日线行情
import_index_daily(
    index_codes=[
        "000001.SH",  # 上证指数
        "399001.SZ",  # 深证成指
        "000300.SH",  # 沪深300
        "000905.SH",  # 中证500
        "399006.SZ",  # 创业板指
        "000688.SH",  # 科创50
        "000016.SH",  # 上证50
        "399005.SZ",  # 中小100
        "000852.SH",  # 中证1000
    ],
    start_date="2010-01-01",
    end_date="2025-12-31",
)

---

一键全量导入 (可选)

如果不想逐步执行，可以运行下面这个 Cell 一键完成所有导入

In [ ]:

# 一键全量导入 (需数小时~数十小时，请谨慎)
# full_import(
#     start_date="2010-01-01",
#     end_date="2025-12-31",
#     import_financials=True,  # 设为 False 跳过财务数据加快速度
# )

---

数据验证与查询示例

In [ ]:

import pandas as pd
import psycopg2

conn = get_pg_connection()

# 各表统计
tables = ["stock_basic", "daily", "daily_basic", "adj_factor",
          "income", "balancesheet", "cashflow", "fina_indicator",
          "trade_cal", "index_daily"]

print(f"{'表名':<20} {'记录数':>12} {'最早日期':>12} {'最晚日期':>12}")
print("-" * 60)
for table in tables:
    try:
        count_sql = f"SELECT COUNT(*) FROM {table}"
        count = pd.read_sql(count_sql, conn).iloc[0, 0]
        
        # 尝试获取日期范围
        date_col = None
        if table == "daily":
            date_col = "trade_date"
        elif table == "daily_basic":
            date_col = "trade_date"
        elif table in ["income", "balancesheet", "cashflow"]:
            date_col = "end_date"
        elif table == "fina_indicator":
            date_col = "end_date"
        elif table == "trade_cal":
            date_col = "cal_date"
        elif table == "index_daily":
            date_col = "trade_date"
        elif table == "adj_factor":
            date_col = "trade_date"
            
        if date_col:
            date_sql = f"SELECT MIN({date_col}), MAX({date_col}) FROM {table}"
            min_d, max_d = pd.read_sql(date_sql, conn).iloc[0]
            print(f"{table:<20} {count:>12,} {str(min_d)[:10]:>12} {str(max_d)[:10]:>12}")
        else:
            print(f"{table:<20} {count:>12,}")
    except Exception as e:
        print(f"{table:<20} {'错误':>12}: {str(e)[:40]}")

conn.close()

In [ ]:

# 示例查询 1: 查询某股票最近10个交易日数据
query1 = """
SELECT trade_date, open, high, low, close, vol, amount, pct_chg
FROM daily
WHERE ts_code = '000001.SZ'
ORDER BY trade_date DESC
LIMIT 10
"""
conn = get_pg_connection()
df1 = pd.read_sql(query1, conn)
print("平安银行(000001.SZ) 最近10个交易日:")
display(df1)
conn.close()

In [ ]:

# 示例查询 2: 日线行情 + 估值指标联合查询 (使用视图)
query2 = """
SELECT *
FROM v_daily_with_valuation
WHERE ts_code = '000001.SZ'
  AND trade_date >= '2024-01-01'
ORDER BY trade_date DESC
LIMIT 10
"""
conn = get_pg_connection()
df2 = pd.read_sql(query2, conn)
print("平安银行 - 日线+估值:")
display(df2[['trade_date', 'close', 'pct_chg', 'pe', 'pe_ttm', 'pb', 'total_mv']])
conn.close()

In [ ]:

# 示例查询 3: 最新财务指标 Top 20 (按 ROE 排序)
query3 = """
SELECT *
FROM v_latest_financials
WHERE roe IS NOT NULL
  AND roe > 0
ORDER BY roe DESC
LIMIT 20
"""
conn = get_pg_connection()
df3 = pd.read_sql(query3, conn)
print("ROE Top 20:")
display(df3[['ts_code', 'name', 'industry', 'roe', 'roa', 'eps', 'debt_to_assets']])
conn.close()