年化210%，回撤10%，夏普3.74的季度榜策略 | akshare下载沪深300成份股到hdf5的python代码

import akshare as akimport pandas as pdimport numpy as npfrom datetime import datetime, timedeltaimport os

def get_stock_data(symbol, name):    """    从akshare获取单个股票的历史数据并预处理    """    print(f"正在获取{name}({symbol})历史数据...")    try:        # 获取股票历史数据        df = ak.stock_zh_a_hist(symbol=symbol, period="daily", start_date="20000101",                                end_date=datetime.now().strftime('%Y%m%d'))
        # 数据预处理        df.rename(columns={            '日期': 'date',            '开盘': 'open',            '收盘': 'close',            '最高': 'high',            '最低': 'low',            '成交量': 'volume',            '成交额': 'amount',            '振幅': 'amplitude',            '涨跌幅': 'pct_change',            '涨跌额': 'change',            '换手率': 'turnover'        }, inplace=True)
        df['date'] = pd.to_datetime(df['date'])  # 转换日期格式        df['symbol'] = symbol        df['name'] = name
        return df    except Exception as e:        print(f"获取股票{symbol}数据失败: {e}")        return pd.DataFrame()

def update_all_stocks_data(store, stock_list, batch_size=100):    """    批量更新所有股票数据到统一的表格中    """    all_data = []
    # 分批处理股票，避免内存溢出    for i in range(0, len(stock_list), batch_size):        batch = stock_list[i:i + batch_size]        batch_data = []
        for stock in batch:            df = get_stock_data(stock, stock)            if not df.empty:                batch_data.append(df)
        if batch_data:            all_data.extend(batch_data)
        print(f"已处理 


    
{min(i + batch_size, len(stock_list))}/{len(stock_list)} 只股票")
    if all_data:        # 合并所有数据        combined_df = pd.concat(all_data, ignore_index=True)
        # 读取现有数据        try:            existing_data = store.select('all_stocks')            # 合并新旧数据，去除重复            combined_df = pd.concat([existing_data, combined_df])            combined_df.drop_duplicates(subset=['symbol', 'date'], keep='last', inplace=True)        except:            pass  # 如果没有现有数据，直接使用新数据
        # 按日期和股票代码排序        combined_df.sort_values(by=['date', 'symbol'], inplace=True)
        # 写回HDF5，创建索引以加速查询        store.put('all_stocks', combined_df, format='table',                  data_columns=['symbol', 'date', 'close'],                  index=['symbol', 'date'])
        return combined_df    else:        return pd.DataFrame()

def create_index_table(store):    """    创建股票代码索引表，便于快速查找特定股票    """    try:        all_data = store.select('all_stocks')        # 获取每个股票的最新数据日期        latest_dates = all_data.groupby('symbol')['date'].max().reset_index()        latest_dates.rename(columns={'date': 'last_update'}, inplace=True)
        # 存储索引表        store.put('stock_index', latest_dates, format='table', data_columns=['symbol'])
        return latest_dates    except:        return pd.DataFrame()

def query_stocks(store, condition):    """    执行条件查询    """    try:        # 使用HDF5的查询功能        result = store.select('all_stocks', where=condition)        return result    except Exception as e:        print(f"查询失败: {e}")        return pd.DataFrame()
def get_csi300_constituents():    """获取沪深300成分股"""    df = ak.index_stock_cons_sina(symbol="000300")    # 统一格式：只保留股票代码（去掉市场前缀）    df["code"] = df["code"].str.replace(r"sh\.|sz\.", "", regex=True)    return df["code"].tolist()
def main():    # 定义HDF5文件路径    hdf5_path = 'all_stocks_data.h5'
    # 示例股票列表 (实际应用中应该从文件或数据库获取)    # stock_list = [    #     {'symbol': '000001', 'name': '平安银行'},    #     {'symbol': '000002', 'name': '万科A'},    #     {'symbol': '000004', 'name': '国华网安'},    #     # ... 这里应该有5000只股票的列表    # ]    stock_list = get_csi300_constituents()

    # 为了演示，我们只使用少量股票    #demo_stock_list = stock_list[:10]
    # 创建或更新HDF5文件    with pd.HDFStore(hdf5_path, mode='a'


    
, complib='blosc', complevel=9) as store:        # 更新所有股票数据        all_data = update_all_stocks_data(store, stock_list)        print(f"所有股票数据已更新，总记录数: {len(all_data)}")
        # 创建索引表        index_table = create_index_table(store)        print("股票索引表已创建")
        # 示例查询 1: close > 10 and close < 20        print("\n查询 close > 10 and close < 20 的股票:")        result1 = query_stocks(store, 'close > 10 & close < 20')        print(f"找到 {len(result1)} 条记录")        if not result1.empty:            print(result1[['symbol', 'name', 'date', 'close']].head())
        # 示例查询 2: 涨幅大于5%的股票        print("\n查询涨幅大于5%的股票:")        result2 = query_stocks(store, 'pct_change > 5')        print(f"找到 {len(result2)} 条记录")        if not result2.empty:            print(result2[['symbol', 'name', 'date', 'pct_change']].head())
        # 示例查询 3: 特定日期的所有股票数据        latest_date = all_data['date'].max()        print(f"\n查询最新交易日 {latest_date.date()} 的所有股票数据:")        result3 = query_stocks(store, f'date == "{latest_date}"')        print(f"找到 {len(result3)} 条记录")        if not result3.empty:            print(result3[['symbol', 'name', 'close', 'pct_change']].head())
        # 示例查询 4: 特定股票的历史数据        print(f"\n查询平安银行(000001)的历史数据:")        result4 = query_stocks(store, 'symbol == "000001"')        print(f"找到 {len(result4)} 条记录")        if not result4.empty:            print(result4[['date', 'close', 'pct_change']].head())
    print(f"\n所有操作已完成，数据已保存到: {hdf5_path}")

if __name__ == "__main__":    main()