指数RPS强度排名分析工具：一键掌握市场板块强弱动向

import streamlit as st
import pywencai
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
# 设置页面配置
st.set_page_config(page_title="指数RPS强度排名", layout="wide")
st.title("指数RPS强度排名分析")
# 日期计算函数
def get_date_range(days):
    end_date = datetime.now()
    start_date = end_date - timedelta(days=days)
    return start_date.strftime("%Y%m%d"), end_date.strftime("%Y%m%d")
# RPS计算函数 - 使用公式: RPS = (1 - 排名 / 总板块数) × 100
def calculate_rps(df, change_col):
    # 转换涨跌幅为数值
    df[change_col] = pd.to_numeric(df[change_col].astype(str).str.replace('%', ''), errors='coerce')
    # 计算排名（按涨跌幅降序排列）
    df['rank'] = df[change_col].rank(ascending=False, method='min')
    # 计算RPS
    total_count = len(df)
    df['RPS'] = ((1 - df['rank'] / total_count) * 100).round(2)
    # 删除临时列
    df.drop('rank', axis=1, inplace=True)
    return df
# 获取数据函数
def get_index_data(period):
    start_date, end_date = get_date_range(period)
    query = f"指数代码886开头，近{period}日涨跌幅"
    try:
        # 使用pywencai获取数据
        df = pywencai.get(query=query, query_type='zhishu')
        # 检查返回数据
        if df.empty:
            st.warning(f"未获取到近{period}日数据")
            return None
        # 打印列名以便调试
        #st.write(f"近{period}日数据列名: {df.columns.tolist()}")
        # 查找涨跌幅列
        change_col = None
        for col in df.columns:
            if "区间涨跌幅" in col:
                change_col = col
                break
        if not change_col:
            st.warning(f"未找到近{period}日涨跌幅列")
            return None
        # 查找代码列
        code_cols = [col for col in df.columns if "指数代码" in col]
        if not code_cols:
            st.warning(f"未找到近{period}日代码列")
            return None
        code_col = code_cols[0]
        # 查找名称列
        name_cols = [col for col in df.columns if "指数简称" in col]
        if not name_cols:
            st.warning(f"未找到近{period}日名称列")
            return None
        name_col = name_cols[0]
        # 提取关键列
        result_df = df[[code_col, name_col, change_col]].copy()
        result_df.columns = ['指数代码', '指数简称', f'{period}日涨跌幅']
        # 计算RPS
        result_df = calculate_rps(result_df, f'{period}日涨跌幅')
        result_df.rename(columns={'RPS': f'RPS_{period}'}, inplace=True)
        return result_df
    except Exception as e:
        st.error(f"获取近{period}日数据失败: {str(e)}")
        return None
# 主程序
def app():
    # 时间范围选择
    periods = st.multiselect(
        "选择时间范围",
        [5, 20, 60],
        default=[5, 20, 60]
    )
    # 数据获取按钮
    if st.button("获取数据"):
        with st.spinner("正在获取数据..."):
            dataframes = {}
            for period in periods:
                df = get_index_data(period)
                if df is not None:
                    dataframes[period] = df
            if not dataframes:
                st.error("未获取到任何数据")
                return
            # 合并数据
            merged_df = None
            for i, period in enumerate(periods):
                if i == 0:
                    merged_df = dataframes[period]
                else:
                    merged_df = pd.merge(
                        merged_df,
                        dataframes[period],
                        on=['指数代码', '指数简称'],
                        how='outer'
                    )
            # 显示结果
            st.subheader(f"指数RPS强度排名 ({', '.join(map(str, periods))}日)")
            st.dataframe(
                merged_df.sort_values(by=[f'RPS_{p}' for p in periods], ascending=False),
                use_container_width=True,
                height=600
            )
            # 下载按钮
            csv = merged_df.to_csv(index=False).encode('utf-8-sig')
            st.download_button(
                label="下载CSV数据",
                data=csv,
                file_name=f"指数RPS排名_{datetime.now().strftime('%Y%m%d')}.csv",
                mime='text/csv'
            )
# 运行主程序
if __name__ == "__main__":
    app()