#!/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()