基于streamlit的多策略多基准可视化分析（代码+数据下载）

关于可视化，这段时间折腾了matplotlib：mplfinance实现K线图，多指标绘图, bokeh等等。backtrader本身用的是matplotlib，而有人帮它实现了bokeh版本的plotting。backtesting框架使用的绘图就是bokeh。

这是backtesting绘的图，效果也还行，bokeh的代码还算简洁，

在多图与布局方面，bokeh比mplfinance方便。

当然，随着机器学习、数据科学的发展，技术总是日新月异。

之前我们使用过streamlit，最近在很多chatGPT的应用上还看到gradio这样的框架。综合比较下来，还是决定再使用stramlit试试。

我们的绘图框架主要做两件事：

一是回测结果的可视化，结果指标，equity曲线，还有交易记录。

二是证券的相关性分析，就是你要把某一个股票纳入投资池，可以通过这个系统看看它的风险、收益特征。可以计算某个因子，把因子可视化出来。

通过上面的描述，如果要展示表格，以及动态交易，那streamlit比bokeh会方便很多。同时，streamlit支持显示bokeh和plotly的绘图的。

先看效果：

可下拉多选策略进行对比：

可以选择日期范围：

可以选择N个比较基准：

关键实现代码如下：（请星友们前往星球下载更新）

注意：不是直接run，而是在terminal里使用streamlit run main_streamlit.py

from datetime import datetime

import pandas as pd
import numpy as np
import os
from engine.config import DATA_H5, DATA_RESULTS

# from engine.logic import Logic
from engine.datafeed.dataloader import Hdf5Dataloader


def universe_symbols():
    with pd.HDFStore(DATA_H5.resolve()) as s:
        return [s.replace('/', '') for s in s.keys()]


def backtest_results():
    files = os.listdir(DATA_RESULTS.resolve())
    h5_files = []
    for f in files:
        if '.h5' in f:
            h5_files.append(f)
    return h5_files


symbols = universe_symbols()
bkt_results = backtest_results()

import streamlit as st

title = '回测结果分析'
st.set_page_config(page_title=title, page_icon=":bar_chart:", layout="wide")
st.title(title)

date_start = st.sidebar.date_input('起始日期：', datetime(2012, 1, 1))
date_start = date_start.strftime('%Y-%m-%d')
date_end = st.sidebar.date_input('结束日期：', datetime.now().date())
date_end = date_end.strftime('%Y-%m-%d')

st.sidebar.header('请选择策略')
stras = st.sidebar.multiselect('请选择策略：', options=bkt_results, default=bkt_results[0])

st.sidebar.header("请选择基准代码:")
symbols = st.sidebar.multiselect(
    "选择基准代码:",
    options=symbols,
    default=['SPX', '000300.SH']
)

loader = Hdf5Dataloader(symbols, start_date=date_start, end_date=date_end)
df = loader.load()
df_returns = df.pivot_table(columns='symbol', values='return_0', index=df.index)


# print(df_returns)


def get_df_stras():
    df_stras_return = []
    for f in stras:
        with pd.HDFStore(DATA_RESULTS.joinpath(f)) as s:
            df = s['portfolio_df']
            se = df['market_value'].pct_change()
            se.name = f.replace('.h5', '')
            df_stras_return.append(se)
    return pd.concat(df_stras_return, axis=1)


df_stras = get_df_stras()
df_returns = pd.concat([df_returns, df_stras], axis=1)
df_returns.dropna(inplace=True)
df_returns = df_returns[df_returns.index >= date_start]
df_returns = df_returns[df_returns.index <= date_end]
df_equity = (1 + df_returns).cumprod()

st.subheader('折线图')
st.line_chart(df_equity)

col_indicators, col_corr = st.columns(2)
col_indicators.subheader('风险收益指标')


def calc_indicators(df_returns):
    print(df_returns.index[0], df_returns.index[-1])
    import empyrical
    accu_returns = empyrical.cum_returns_final(df_returns)
    accu_returns.name = '累计收益'
    annu_returns = empyrical.annual_return(df_returns)
    annu_returns.name = '年化收益'
    max_drawdown = empyrical.max_drawdown(df_returns)
    max_drawdown.name = '最大回撤'
    sharpe = empyrical.sharpe_ratio(df_returns)
    sharpe = pd.Series(sharpe)
    sharpe.name = '夏普比'
    max_drawdown.index = annu_returns.index

    sharpe = pd.Series(empyrical.sharpe_ratio(df_returns))
    sharpe.index = accu_returns.index
    sharpe.name = '夏普比'
    all = pd.concat([accu_returns, annu_returns, max_drawdown, sharpe], axis=1)
    print(all)
    return all


col_indicators.table(calc_indicators(df_returns))

col_corr.subheader('相关系数')
col_corr.table(df_returns.corr())