【Python技术】怎么管理自己的问财多策略

import streamlit as st
import pandas as pd
import json
import os
import pywencai
# 初始化Session State
if 'strategies' not in st.session_state:
    st.session_state.strategies = {}
if 'last_result' not in st.session_state:
    st.session_state.last_result = None
# 策略存储文件名
STRATEGY_FILE = "wencai_strategies.json"
# 加载策略
def load_strategies():
    if os.path.exists(STRATEGY_FILE):
        with open(STRATEGY_FILE, 'r') as f:
            st.session_state.strategies = json.load(f)
# 保存策略
def save_strategies():
    with open(STRATEGY_FILE, 'w') as f:
        json.dump(st.session_state.strategies, f, indent=2)
# 执行问财查询
def run_wencai_query(query_str):
    """执行问财查询并返回结果"""
    try:
        if not query_str:
            st.warning("查询条件不能为空")
            return None
        # 使用pywencai执行查询
        result = pywencai.get(query=query_str)
        return result
    except Exception as e:
        st.error(f"查询失败: {str(e)}")
        return None
# 页面布局
st.set_page_config(
    page_title="问财策略查询工具",
    page_icon="📊",
    layout="wide"
)
# 加载策略
load_strategies()
# 侧边栏 - 策略管理
with st.sidebar:
    st.header("📋 策略管理")
    # 添加新策略
    with st.expander("➕ 添加新策略", expanded=False):
        with st.form("add_strategy"):
            strategy_name = st.text_input("策略名称", key="strategy_name")
            condition = st.text_area("完整查询语句", key="condition_input",
                                     help="例如: ROE连续三年大于10%，净利率同比增长超过20%，市盈率小于30")
            submit = st.form_submit_button("保存策略")
            if submit and strategy_name:
                if strategy_name in st.session_state.strategies:
                    st.warning("该策略名称已存在！")
                elif not condition.strip():
                    st.warning("请输入查询语句")
                else:
                    st.session_state.strategies[strategy_name] = {
                        "query": condition.strip(),
                        "last_used": pd.Timestamp.now().strftime("%Y-%m-%d %H:%M:%S")
                    }
                    save_strategies()
                    st.success(f"策略 '{strategy_name}' 已保存！")
    # 策略列表
    st.divider()
    st.subheader("已保存策略")
    if not st.session_state.strategies:
        st.info("暂无策略，请添加新策略")
    else:
        for name, details in list(st.session_state.strategies.items()):
            col1, col2 = st.columns([0.7, 0.3])
            with col1:
                st.markdown(f"**{name}**")
                st.caption(f"最后使用: {details['last_used']}")
            with col2:
                if st.button("删除", key=f"del_{name}"):
                    del st.session_state.strategies[name]
                    save_strategies()
                    st.rerun()
# 主面板
st.title("📊 问财策略查询")
# 策略选择和查询执行
if st.session_state.strategies:
    # 策略选择
    selected_strategy = st.selectbox(
        "选择策略",
        list(st.session_state.strategies.keys()),
        index=0
    )
    # 显示策略详情
    strategy_details = st.session_state.strategies[selected_strategy]
    query_str = strategy_details["query"]
    st.subheader("策略详情")
    st.info(f"**策略名称**: {selected_strategy}")
    st.code(query_str, language="sql")
    # 编辑策略按钮
    with st.expander("✏️ 编辑策略", expanded=False):
        new_query = st.text_area("修改查询语句", value=query_str, height=150)
        if st.button("💾 保存修改", key=f"update_{selected_strategy}"):
            st.session_state.strategies[selected_strategy]["query"] = new_query.strip()
            save_strategies()
            st.success("策略已更新！")
            st.rerun()
    # 执行查询
    if st.button("🚀 执行查询", type="primary", key="run_query"):
        with st.spinner(f"执行中... 正在查询 {selected_strategy}"):
            # 更新最后使用时间
            st.session_state.strategies[selected_strategy]["last_used"] = pd.Timestamp.now().strftime(
                "%Y-%m-%d %H:%M:%S")
            save_strategies()
            # 执行查询
            result = run_wencai_query(query_str)
            if result is not None:
                if not result.empty:
                    st.session_state.last_result = result
                    st.session_state.last_strategy = selected_strategy
                    st.session_state.last_query = query_str
                else:
                    st.info("未找到匹配的结果")
# 显示结果区域
if st.session_state.last_result is not None:
    st.divider()
    st.subheader(f"查询结果: {st.session_state.last_strategy}")
    # 显示查询语句
    st.info(f"**查询语句**: {st.session_state.last_query}")
    df = st.session_state.last_result
    # 简单数据预处理
    if '代码' in df.columns:
        df['代码'] = df['代码'].astype(str).str.zfill(6)
    # 结果展示
    st.dataframe(df, height=600)
    # 下载按钮
    csv = df.to_csv(index=False).encode('utf-8')
    st.download_button(
        label="下载CSV",
        data=csv,
        file_name=f"wencai_{st.session_state.last_strategy.replace(' ', '_')}.csv",
        mime='text/csv'
    )
else:
    st.info("请选择策略后执行查询，或使用直接查询功能")