cyb50-quant/a50-futures/fetch_a50_data.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
A50期货 - 30分钟K线数据获取
标的：富时中国A50指数期货 (CN0Y) / 华夏A50ETF (159601) 作为替代
数据周期：近40天

注意：由于网络限制，部分数据源可能无法访问
"""

import pandas as pd
import numpy as np
import akshare as ak
import requests
import json
from datetime import datetime, timedelta
import warnings
warnings.filterwarnings('ignore')


class A50FuturesDataFetcher:
    """A50期货数据获取类"""
    
    def __init__(self):
        self.symbol = "CN0Y"  # A50期货主力合约代码
        print(f"A50期货数据获取器初始化")
        print(f"标的代码: {self.symbol}")
    
    def fetch_30min_data(self, days=40):
        """
        获取A50相关指数30分钟K线数据
        
        Parameters:
            days: 获取近N天的数据（默认40天）
        
        Returns:
            DataFrame with columns: datetime, open, high, low, close, volume
        """
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days+5)
        
        print(f"\n{'='*60}")
        print(f"获取A50相关数据 (近{days}天)")
        print(f"时间范围: {start_date.strftime('%Y-%m-%d')} 至 {end_date.strftime('%Y-%m-%d')}")
        print(f"{'='*60}")
        
        # 尝试多个数据源
        df = None
        
        # 方法1: 使用A50 ETF作为替代
        df = self._fetch_a50_etf(start_date, end_date)
        if df is not None and len(df) > 0:
            print(f"\n✅ 使用A50 ETF数据作为替代")
            return df
        
        # 方法2: 使用上证50指数
        df = self._fetch_sz50_index(start_date, end_date)
        if df is not None and len(df) > 0:
            print(f"\n✅ 使用上证50指数作为替代")
            return df
        
        # 方法3: 生成模拟数据（用于测试）
        print("\n⚠️ 无法获取实时数据，生成模拟数据用于测试...")
        df = self._generate_sample_data(days)
        return df
    
    def _fetch_a50_etf(self, start_date, end_date):
        """获取华夏A50ETF数据作为A50期货的替代"""
        try:
            print("\n[数据源1] 获取华夏A50ETF(159601)数据...")
            
            # 使用akshare获取ETF分钟数据
            df = ak.fund_etf_hist_min_em(symbol="159601", period="30",
                                          start_date=start_date.strftime('%Y%m%d%H%M%S'),
                                          end_date=end_date.strftime('%Y%m%d%H%M%S'))
            
            if df is not None and not df.empty and len(df) > 10:
                # 标准化列名
                df = df.rename(columns={
                    '时间': 'datetime',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume'
                })
                
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                
                print(f"   ✅ 获取成功: {len(df)}条")
                return df
            else:
                print(f"   数据不足")
                
        except Exception as e:
            print(f"   获取失败: {e}")
        
        return None
    
    def _fetch_sz50_index(self, start_date, end_date):
        """获取上证50指数作为替代"""
        try:
            print("\n[数据源2] 获取上证50指数(000016)数据...")
            
            df = ak.index_zh_a_hist_min_em(symbol="000016", period="30",
                                            start_date=start_date.strftime('%Y%m%d%H%M%S'),
                                            end_date=end_date.strftime('%Y%m%d%H%M%S'))
            
            if df is not None and not df.empty and len(df) > 10:
                df = df.rename(columns={
                    '时间': 'datetime',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume'
                })
                
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                
                print(f"   ✅ 获取成功: {len(df)}条")
                return df
            else:
                print(f"   数据不足")
                
        except Exception as e:
            print(f"   获取失败: {e}")
        
        return None
    
    def _generate_sample_data(self, days=40):
        """生成A50风格的模拟数据（用于测试）"""
        print("\n[模拟数据] 生成A50风格模拟数据...")
        
        # 生成时间序列（交易日9:30-11:30, 13:00-15:00）
        dates = pd.date_range(end=datetime.now(), periods=days, freq='B')  # 工作日
        all_times = []
        
        for date in dates:
            # 上午 9:30-11:30 (4个30分钟)
            for hour, minute in [(9, 30), (10, 0), (10, 30), (11, 0)]:
                all_times.append(date.replace(hour=hour, minute=minute))
            # 下午 13:00-15:00 (4个30分钟)
            for hour, minute in [(13, 0), (13, 30), (14, 0), (14, 30)]:
                all_times.append(date.replace(hour=hour, minute=minute))
        
        all_times = sorted([t for t in all_times if t <= datetime.now()])
        n = len(all_times)
        
        # A50基准价约13500点
        base_price = 13500
        
        # 生成随机价格序列（带趋势）
        np.random.seed(42)
        returns = np.random.normal(0.0002, 0.008, n)  # A50波动率
        
        # 添加一些趋势
        trend = np.sin(np.linspace(0, 4*np.pi, n)) * 0.002
        returns += trend
        
        # 计算价格
        prices = base_price * np.exp(np.cumsum(returns))
        
        # 生成OHLC数据
        data = []
        for i, (time, price) in enumerate(zip(all_times, prices)):
            volatility = price * 0.003  # 0.3%波动
            open_price = price + np.random.normal(0, volatility * 0.3)
            high_price = max(open_price, price) + np.random.uniform(0, volatility)
            low_price = min(open_price, price) - np.random.uniform(0, volatility)
            close_price = price
            volume = np.random.randint(100000, 500000)
            
            data.append({
                'datetime': time,
                'open': round(open_price, 2),
                'high': round(high_price, 2),
                'low': round(low_price, 2),
                'close': round(close_price, 2),
                'volume': volume
            })
        
        df = pd.DataFrame(data)
        df = df.set_index('datetime').sort_index()
        
        print(f"   ✅ 生成模拟数据: {len(df)}条")
        print(f"   ⚠️  注意：此为模拟数据，仅用于测试")
        
        return df
    
    def save_to_csv(self, df, filename=None):
        """保存数据到CSV文件"""
        if df is None or len(df) == 0:
            print("❌ 无数据可保存")
            return None
        
        if filename is None:
            filename = f"A50_futures_30min_{datetime.now().strftime('%Y%m%d')}.csv"
        
        try:
            df.to_csv(filename)
            import os
            print(f"\n💾 数据已保存: {filename}")
            print(f"   数据条数: {len(df)}")
            print(f"   文件大小: {round(os.path.getsize(filename)/1024, 2)} KB")
            return filename
        except Exception as e:
            print(f"❌ 保存失败: {e}")
            return None
    
    def show_statistics(self, df):
        """显示数据统计信息"""
        if df is None or len(df) == 0:
            return
        
        print(f"\n{'='*60}")
        print(f"数据统计信息")
        print(f"{'='*60}")
        
        print(f"\n【基本信息】")
        print(f"  数据条数: {len(df)}")
        print(f"  时间区间: {df.index[0]} 至 {df.index[-1]}")
        unique_dates = set([d.date() for d in pd.to_datetime(df.index)])
        print(f"  交易天数: {len(unique_dates)}天")
        
        print(f"\n【价格统计】")
        print(f"  最高价: {df['high'].max():.2f}")
        print(f"  最低价: {df['low'].min():.2f}")
        print(f"  最新价: {df['close'].iloc[-1]:.2f}")
        
        # 计算涨跌幅
        price_change = (df['close'].iloc[-1] - df['close'].iloc[0]) / df['close'].iloc[0] * 100
        print(f"  区间涨跌: {price_change:+.2f}%")
        
        print(f"\n【成交量统计】")
        print(f"  总成交量: {df['volume'].sum():,.0f}")
        print(f"  日均成交: {df['volume'].mean():,.0f}")
        
        print(f"\n【波动率统计】")
        df['returns'] = df['close'].pct_change()
        volatility = df['returns'].std() * np.sqrt(48) * 100  # 年化波动率
        print(f"  年化波动率: {volatility:.2f}%")


def main():
    """主程序"""
    print("="*70)
    print("A50期货30分钟K线数据获取工具")
    print("="*70)
    print(f"执行时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
    
    # 创建数据获取器
    fetcher = A50FuturesDataFetcher()
    
    # 获取近40天数据
    df = fetcher.fetch_30min_data(days=40)
    
    if df is not None and len(df) > 0:
        # 显示统计信息
        fetcher.show_statistics(df)
        
        # 保存到CSV
        fetcher.save_to_csv(df)
        
        # 显示最新10条数据
        print(f"\n{'='*60}")
        print(f"最新10条数据")
        print(f"{'='*60}")
        print(df.tail(10)[['open', 'high', 'low', 'close', 'volume']].to_string())
        
        print(f"\n✅ 数据获取完成!")
    else:
        print(f"\n❌ 数据获取失败")
    
    print("="*70)


if __name__ == "__main__":
    import os
    main()