board-monitor/fetch_history.py

#!/usr/bin/env python3
"""
批量获取板块历史K线数据
"""

import urllib.request
import json
import sqlite3
import os
import time
from datetime import datetime
from pathlib import Path

# 清除代理
for v in ['http_proxy', 'https_proxy', 'HTTP_PROXY', 'HTTPS_PROXY']:
    os.environ.pop(v, None)

SCRIPT_DIR = Path(__file__).parent
DATA_DIR = SCRIPT_DIR / "data"
DB_FILE = DATA_DIR / "board_history.db"


def get_board_codes():
    """从数据库获取已保存的板块代码"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    cursor.execute('SELECT DISTINCT board_code, board_name FROM board_data WHERE board_type="industry"')
    industry = cursor.fetchall()
    
    cursor.execute('SELECT DISTINCT board_code, board_name FROM board_data WHERE board_type="concept"')
    concept = cursor.fetchall()
    
    conn.close()
    return industry, concept


def get_kline(code: str, days: int = 25):
    """获取板块历史K线"""
    secid = f"90.{code}"
    url = f"http://push2his.eastmoney.com/api/qt/stock/kline/get?secid={secid}&fields1=f1,f2,f3,f4,f5,f6&fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61,f62,f63&klt=101&fqt=1&end=20500101&lmt={days}"
    
    try:
        req = urllib.request.Request(url, headers={
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
            'Referer': 'http://quote.eastmoney.com/'
        })
        
        with urllib.request.urlopen(req, timeout=15) as resp:
            data = json.loads(resp.read().decode())
            
            if data.get('data') and data['data'].get('klines'):
                klines = data['data']['klines']
                result = []
                
                for kline in klines:
                    parts = kline.split(',')
                    # 格式: 日期,开盘,收盘,最高,最低,成交量,成交额,振幅,涨跌幅,涨跌额,换手率
                    result.append({
                        'date': parts[0],
                        'pct_change': float(parts[8]) if parts[8] != '-' else 0,
                        'amount': float(parts[6]) if parts[6] != '-' else 0,
                    })
                
                return result
    except Exception as e:
        pass
    
    return None


def save_to_db(board_type: str, code: str, name: str, klines: list):
    """保存到数据库"""
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    count = 0
    for kline in klines:
        try:
            cursor.execute('''
                INSERT OR REPLACE INTO board_data 
                (date, board_type, board_code, board_name, pct_change, main_flow, leader_name)
                VALUES (?, ?, ?, ?, ?, ?, ?)
            ''', (
                kline['date'],
                board_type,
                code,
                name,
                kline['pct_change'],
                kline['amount'] / 1e8,
                ''
            ))
            count += 1
        except:
            continue
    
    conn.commit()
    conn.close()
    return count


def main():
    print("\n📊 批量获取板块历史数据")
    print("=" * 50)
    
    industry, concept = get_board_codes()
    print(f"\n行业板块: {len(industry)} 个")
    print(f"概念板块: {len(concept)} 个")
    
    # 获取行业板块历史
    print("\n📡 获取行业板块历史数据...")
    success = 0
    for i, (code, name) in enumerate(industry, 1):
        print(f"  [{i}/{len(industry)}] {name}...", end=' ', flush=True)
        
        klines = get_kline(code, 25)
        if klines:
            count = save_to_db('industry', code, name, klines)
            print(f"✅ {count}条")
            success += 1
        else:
            print("❌")
        
        time.sleep(0.15)
    
    print(f"\n行业板块完成: {success}/{len(industry)}")
    
    # 获取概念板块历史
    print("\n📡 获取概念板块历史数据...")
    success = 0
    for i, (code, name) in enumerate(concept, 1):
        print(f"  [{i}/{len(concept)}] {name}...", end=' ', flush=True)
        
        klines = get_kline(code, 25)
        if klines:
            count = save_to_db('concept', code, name, klines)
            print(f"✅ {count}条")
            success += 1
        else:
            print("❌")
        
        time.sleep(0.15)
    
    print(f"\n概念板块完成: {success}/{len(concept)}")
    
    # 统计
    conn = sqlite3.connect(DB_FILE)
    cursor = conn.cursor()
    
    cursor.execute('SELECT COUNT(DISTINCT date) FROM board_data')
    date_count = cursor.fetchone()[0]
    
    cursor.execute('SELECT MIN(date), MAX(date) FROM board_data')
    min_date, max_date = cursor.fetchone()
    
    cursor.execute('SELECT COUNT(*) FROM board_data')
    total = cursor.fetchone()[0]
    
    cursor.execute('SELECT date, COUNT(*) FROM board_data GROUP BY date ORDER BY date')
    dates = cursor.fetchall()
    
    conn.close()
    
    print(f"\n✅ 完成!")
    print(f"   数据日期: {date_count} 天")
    print(f"   日期范围: {min_date} ~ {max_date}")
    print(f"   总记录: {total} 条")
    
    print(f"\n   日期分布:")
    for date, cnt in dates[-10:]:
        print(f"   {date}: {cnt} 条")


if __name__ == "__main__":
    main()