cyb50-quant/cat-fly/fetch_mairui_cyb50.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
麦儒数据 (mairui) API - 获取创业板50指数30分钟K线数据
文档: https://www.mairui.club/
Token: AE17EE23-AAE4-492F-A959-EC883DFA5A76
"""

import pandas as pd
import requests
import json
import os
from datetime import datetime, timedelta
import time
import warnings
warnings.filterwarnings('ignore')


class MairuiDataFetcher:
    """麦儒数据获取器"""

    def __init__(self):
        self.token = "AE17EE23-AAE4-492F-A959-EC883DFA5A76"
        self.base_url = "https://api.mairuiapi.com"
        self.data_dir = "data"
        os.makedirs(self.data_dir, exist_ok=True)

    def get_index_minute_data(self, index_code="399673.SZ", period="30", start_date=None, end_date=None):
        """
        获取指数分钟历史数据
        :param index_code: 指数代码，创业板50 = 399673.SZ
        :param period: 周期，支持 1,5,15,30,60,d(日线)
        :param start_date: 开始时间 (YYYYMMDD)
        :param end_date: 结束时间 (YYYYMMDD)
        :return: DataFrame
        """
        try:
            # 麦儒API接口：获取指数历史K线数据
            # 格式: /hsindex/history/指数代码/周期/Token?st=开始时间&et=结束时间
            url = f"{self.base_url}/hsindex/history/{index_code}/{period}/{self.token}"

            # 添加时间参数
            params = {}
            if start_date:
                params['st'] = start_date
            if end_date:
                params['et'] = end_date

            if params:
                url += "?" + "&".join([f"{k}={v}" for k, v in params.items()])

            print(f"[麦儒数据] 请求URL: {url}")

            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }

            response = requests.get(url, headers=headers, timeout=30)
            response.encoding = 'utf-8'

            if response.status_code != 200:
                print(f"[错误] HTTP状态码: {response.status_code}")
                return None

            data = response.json()

            if isinstance(data, dict) and 'error' in data:
                print(f"[错误] API返回错误: {data['error']}")
                return None

            if not data or not isinstance(data, list):
                print(f"[错误] 返回数据格式不正确")
                return None

            # 转换为DataFrame
            df = pd.DataFrame(data)

            # 标准化列名
            column_mapping = {
                'd': 'DateTime',  # 日期时间
                'o': 'Open',      # 开盘
                'h': 'High',      # 最高
                'l': 'Low',       # 最低
                'c': 'Close',     # 收盘
                'v': 'Volume',    # 成交量
                'e': 'Amount',    # 成交额
                't': 'DateTime'   # 有些接口用t表示时间
            }

            df.rename(columns=column_mapping, inplace=True)

            # 转换时间格式
            if 'DateTime' in df.columns:
                df['DateTime'] = pd.to_datetime(df['DateTime'])
                df.set_index('DateTime', inplace=True)
                df.sort_index(inplace=True)

            # 转换数值类型
            numeric_cols = ['Open', 'High', 'Low', 'Close', 'Volume', 'Amount']
            for col in numeric_cols:
                if col in df.columns:
                    df[col] = pd.to_numeric(df[col], errors='coerce')

            print(f"[麦儒数据] 成功获取 {len(df)} 条数据")
            print(f"[麦儒数据] 时间范围: {df.index[0]} 至 {df.index[-1]}")

            return df

        except Exception as e:
            print(f"[错误] 获取数据失败: {e}")
            return None

    def get_index_history(self, index_code="399673"):
        """
        获取指数历史数据（日线）
        :param index_code: 指数代码
        :return: DataFrame
        """
        try:
            # 麦儒API接口：获取指数历史数据
            url = f"{self.base_url}/zslsh/{index_code}/{self.token}"

            print(f"[麦儒数据] 获取历史数据: {url}")

            headers = {
                'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36'
            }

            response = requests.get(url, headers=headers, timeout=30)
            response.encoding = 'utf-8'

            if response.status_code != 200:
                print(f"[错误] HTTP状态码: {response.status_code}")
                return None

            data = response.json()

            if not data or not isinstance(data, list):
                print(f"[错误] 返回数据格式不正确")
                return None

            df = pd.DataFrame(data)

            # 标准化列名
            column_mapping = {
                'd': 'DateTime',
                'o': 'Open',
                'h': 'High',
                'l': 'Low',
                'c': 'Close',
                'v': 'Volume',
                'e': 'Amount',
                'zf': 'Change_Pct',  # 涨跌幅
                'zde': 'Change_Amount'  # 涨跌额
            }

            df.rename(columns=column_mapping, inplace=True)

            if 'DateTime' in df.columns:
                df['DateTime'] = pd.to_datetime(df['DateTime'])
                df.set_index('DateTime', inplace=True)
                df.sort_index(inplace=True)

            return df

        except Exception as e:
            print(f"[错误] 获取历史数据失败: {e}")
            return None

    def get_cyb50_30min_2023_to_now(self):
        """
        获取创业板50指数2023年至今的30分钟数据
        """
        print("=" * 70)
        print("麦儒数据 - 创业板50指数30分钟数据获取")
        print("=" * 70)

        # 设置时间范围
        start_date = "20230101"
        end_date = datetime.now().strftime("%Y%m%d")

        print(f"时间范围: {start_date} 至 {end_date}")

        # 获取30分钟数据
        data = self.get_index_minute_data(
            index_code="399673.SZ",
            period="30",
            start_date=start_date,
            end_date=end_date
        )

        if data is not None and not data.empty:
            # 筛选2023年至今的数据
            start_date = pd.to_datetime("2023-01-01")
            data = data[data.index >= start_date]

            print(f"\n筛选后数据: {len(data)} 条")
            print(f"时间范围: {data.index[0]} 至 {data.index[-1]}")

            return data

        return None

    def save_data(self, data, filename=None):
        """保存数据到CSV和TXT格式"""
        if data is None or data.empty:
            print("[错误] 没有数据可保存")
            return None

        if filename is None:
            timestamp = datetime.now().strftime("%Y%m%d")
            filename = f"cyb50_30min_2023_to_{timestamp}.csv"

        filepath = os.path.join(self.data_dir, filename)

        # 保存CSV
        data.to_csv(filepath, encoding='utf-8-sig')

        # 保存TXT（兼容原有格式）
        txt_filename = filename.replace('.csv', '.txt')
        txt_filepath = os.path.join(self.data_dir, txt_filename)

        with open(txt_filepath, 'w', encoding='utf-8') as f:
            f.write("创业板50指数 30分钟数据 (来源:麦儒数据)\n")
            f.write("Date Time Open High Low Close Volume Amount\n")
            for idx, row in data.iterrows():
                date_str = idx.strftime('%Y/%m/%d')
                time_str = idx.strftime('%H%M')
                f.write(f"{date_str} {time_str} "
                        f"{row['Open']:.2f} {row['High']:.2f} {row['Low']:.2f} {row['Close']:.2f} "
                        f"{row['Volume']:.0f} {row.get('Amount', 0):.2f}\n")

        print(f"\n[保存成功]")
        print(f"  CSV: {filepath}")
        print(f"  TXT: {txt_filepath}")
        print(f"  记录数: {len(data)}")
        print(f"  文件大小: {os.path.getsize(filepath) / 1024:.2f} KB")

        return filepath


def main():
    """主函数"""
    print("\n" + "=" * 70)
    print("  麦儒数据 - 创业板50指数 30分钟K线数据获取")
    print("  时间范围: 2023年至今")
    print("=" * 70 + "\n")

    fetcher = MairuiDataFetcher()

    # 获取数据
    data = fetcher.get_cyb50_30min_2023_to_now()

    if data is not None and not data.empty:
        # 保存数据
        fetcher.save_data(data)

        # 数据统计
        print("\n" + "=" * 70)
        print("数据统计")
        print("=" * 70)
        print(f"总记录数: {len(data)}")
        print(f"交易日数: {len(set(data.index.date))}")
        print(f"时间范围: {data.index[0]} 至 {data.index[-1]}")
        print(f"\n价格统计:")
        print(f"  最高价: {data['High'].max():.2f}")
        print(f"  最低价: {data['Low'].min():.2f}")
        print(f"  最新价: {data['Close'].iloc[-1]:.2f}")
        print(f"  涨跌幅: {(data['Close'].iloc[-1] / data['Close'].iloc[0] - 1) * 100:.2f}%")
        print("=" * 70 + "\n")
    else:
        print("\n[错误] 数据获取失败\n")


if __name__ == "__main__":
    main()