cyb50-quant/a50-futures/fetch_a50_v2.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
A50期货 - 30分钟K线数据获取（多方案尝试）
标的：富时中国A50指数期货 (CN0Y, XIN9)
"""

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


class A50FuturesFetcher:
    """A50期货数据获取器 - 多方案"""
    
    def __init__(self):
        print("="*70)
        print("A50期货30分钟K线数据获取工具")
        print("="*70)
    
    def fetch_data(self, days=40):
        """获取数据 - 尝试多种方案"""
        end_date = datetime.now()
        start_date = end_date - timedelta(days=days+5)
        
        print(f"\n目标: 获取近{days}天30分钟K线")
        print(f"时间范围: {start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')}")
        
        # 方案1: AKShare期货分钟数据
        print("\n" + "-"*70)
        print("[方案1] AKShare期货分钟数据...")
        df = self._try_akshare_futures(start_date, end_date)
        if df is not None: return df
        
        # 方案2: AKShare股指期货
        print("\n" + "-"*70)
        print("[方案2] AKShare股指期货(CN0Y)...")
        df = self._try_akshare_cn0y(start_date, end_date)
        if df is not None: return df
        
        # 方案3: 新浪财经期货
        print("\n" + "-"*70)
        print("[方案3] 新浪财经期货接口...")
        df = self._try_sina_futures(days)
        if df is not None: return df
        
        # 方案4: 东方财富期货
        print("\n" + "-"*70)
        print("[方案4] 东方财富期货接口...")
        df = self._try_eastmoney_futures()
        if df is not None: return df
        
        # 方案5: 使用相关ETF作为替代
        print("\n" + "-"*70)
        print("[方案5] A50相关ETF数据...")
        df = self._try_a50_etf(start_date, end_date)
        if df is not None: return df
        
        # 方案6: 上证50指数作为替代
        print("\n" + "-"*70)
        print("[方案6] 上证50指数(000016)作为替代...")
        df = self._try_sz50_index(start_date, end_date)
        if df is not None: return df
        
        print("\n❌ 所有方案均失败")
        return None
    
    def _try_akshare_futures(self, start_date, end_date):
        """方案1: AKShare期货分钟数据"""
        try:
            # 尝试获取富时A50期货分钟数据
            print("   尝试: futures_zh_minute_sina...")
            
            # 富时中国A50期货代码
            symbol = "CN0Y"
            df = ak.futures_zh_minute_sina(symbol=symbol, period="30")
            
            if df is not None and len(df) > 0:
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                df = df[(df.index >= start_date) & (df.index <= end_date)]
                
                if len(df) > 10:
                    print(f"   ✅ 成功: {len(df)}条")
                    return df
                    
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def _try_akshare_cn0y(self, start_date, end_date):
        """方案2: AKShare股指期货数据"""
        try:
            print("   尝试: futures_zh_realtime...")
            
            # 获取股指期货实时数据（包含历史）
            df = ak.futures_zh_realtime(symbol="富时A50")
            
            if df is not None and len(df) > 0:
                print(f"   获取到实时数据: {len(df)}条")
                # 注意：这是实时报价，不是K线
                return None
                
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def _try_sina_futures(self, days):
        """方案3: 新浪财经期货K线"""
        try:
            print("   尝试: 新浪财经期货K线接口...")
            
            # 新浪财经A50连续合约
            url = "https://stock.finance.sina.com.cn/futures/api/jsonp.php/var_CN0Y=/InnerFuturesNewService.getMinLine?symbol=CN0Y"
            
            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36',
                'Referer': 'https://finance.sina.com.cn/'
            }
            
            response = requests.get(url, headers=headers, timeout=10)
            
            # 解析JSONP
            text = response.text
            json_start = text.find('[')
            json_end = text.rfind(']') + 1
            
            if json_start > 0 and json_end > json_start:
                json_str = text[json_start:json_end]
                data = json.loads(json_str)
                
                if data and len(data) > 0:
                    df_list = []
                    for item in data:
                        df_list.append({
                            'datetime': item.get('time'),
                            'open': float(item.get('open', 0)),
                            'high': float(item.get('high', 0)),
                            'low': float(item.get('low', 0)),
                            'close': float(item.get('close', 0)),
                            'volume': float(item.get('volume', 0))
                        })
                    
                    df = pd.DataFrame(df_list)
                    df['datetime'] = pd.to_datetime(df['datetime'])
                    df = df.set_index('datetime').sort_index()
                    
                    # 只保留近N天
                    start = datetime.now() - timedelta(days=days)
                    df = df[df.index >= start]
                    
                    if len(df) > 10:
                        print(f"   ✅ 成功: {len(df)}条")
                        return df
                        
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def _try_eastmoney_futures(self):
        """方案4: 东方财富期货数据"""
        try:
            print("   尝试: 东方财富期货接口...")
            
            # 东方财富A50期货
            # 使用股指期货分钟数据接口
            df = ak.futures_zh_minute_sina(symbol="XIN9", period="30")
            
            if df is not None and len(df) > 0:
                df['datetime'] = pd.to_datetime(df['datetime'])
                df = df.set_index('datetime').sort_index()
                
                if len(df) > 10:
                    print(f"   ✅ 成功: {len(df)}条")
                    return df
                    
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def _try_a50_etf(self, start_date, end_date):
        """方案5: A50 ETF数据"""
        try:
            print("   尝试: 华夏A50ETF(159601)...")
            
            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 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)}条 (ETF替代数据)")
                return df
                
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def _try_sz50_index(self, start_date, end_date):
        """方案6: 上证50指数"""
        try:
            print("   尝试: 上证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 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)}条 (上证50替代数据)")
                return df
                
        except Exception as e:
            print(f"   失败: {str(e)[:80]}")
        return None
    
    def save_and_show(self, df, filename=None):
        """保存并显示数据"""
        if df is None or len(df) == 0:
            print("\n❌ 无数据")
            return
        
        # 保存
        if filename is None:
            filename = f"A50_30min_{datetime.now().strftime('%Y%m%d_%H%M')}.csv"
        
        df.to_csv(filename)
        print(f"\n💾 已保存: {filename}")
        
        # 统计
        print(f"\n📊 数据统计:")
        print(f"   数据条数: {len(df)}")
        print(f"   时间区间: {df.index[0]} ~ {df.index[-1]}")
        print(f"   价格区间: {df['low'].min():.2f} ~ {df['high'].max():.2f}")
        print(f"   最新价格: {df['close'].iloc[-1]:.2f}")
        
        # 显示最新10条
        print(f"\n📋 最新10条数据:")
        print(df.tail(10)[['open', 'high', 'low', 'close', 'volume']].to_string())


def main():
    fetcher = A50FuturesFetcher()
    df = fetcher.fetch_data(days=40)
    
    if df is not None:
        fetcher.save_and_show(df)
        print("\n✅ 完成!")
    else:
        print("\n❌ 获取失败，请检查网络连接")
    
    print("="*70)


if __name__ == "__main__":
    main()