app.py
"""
QTrading Web GUI 应用模块
"""
import dash
from dash import dcc, html, Input, Output, State, dash_table
import dash_bootstrap_components as dbc
import pandas as pd
from typing import List, Dict, Optional
from data.data_manager import DataManager
from engine.selector_engine import SelectorEngine
from backtest.backtest_engine import BacktestEngine
from strategy.strategy_manager import StrategyManager
from config.config import config, ADJUST_METHOD
from utils.logger import get_logger
logger = get_logger(__name__)
# 初始化应用
app = dash.Dash(
__name__,
external_stylesheets=[dbc.themes.BOOTSTRAP],
suppress_callback_exceptions=True,
title="QTrading 量化选股系统"
)
app.layout = dbc.Container(
fluid=True,
children=[
# 导航栏
dbc.Navbar(
[
html.A(
dbc.Row(
[
dbc.Col(html.Img(src="", height="30px")),
dbc.Col(dbc.NavbarBrand("QTrading 量化选股系统", className="ms-2")),
],
align="center",
className="g-0",
),
href="#",
style={"textDecoration": "none"},
),
],
color="dark",
dark=True,
className="mb-4",
),
# 主内容区域
dbc.Row([
# 侧边栏
dbc.Col(
width=3,
children=[
dbc.Card(
[
dbc.CardHeader("参数配置"),
dbc.CardBody([
# 策略选择
html.H6("策略选择"),
dcc.Dropdown(
id="strategy-selector",
options=[],
value=config.strategy.default_strategy,
clearable=False,
className="mb-3",
),
# 复权方式
html.H6("复权方式"),
dcc.Dropdown(
id="adjust-method-selector",
options=[
{"label": value, "value": key}
for key, value in ADJUST_METHOD.items()
],
value="qfq",
clearable=False,
className="mb-3",
),
# 选股数量
html.H6("选股数量"),
dcc.Slider(
id="top-n-selector",
min=5,
max=50,
step=5,
value=10,
marks={i: str(i) for i in range(5, 51, 5)},
className="mb-3",
),
# 操作按钮
dbc.ButtonGroup(
[
dbc.Button(
"开始选股",
id="start-select-btn",
color="primary",
size="lg",
className="me-2",
),
dbc.Button(
"更新数据",
id="update-data-btn",
color="secondary",
size="lg",
),
],
className="w-100",
),
# 回测配置
html.Hr(),
html.H6("回测配置"),
dbc.Input(
id="backtest-start-date",
type="text",
placeholder="开始日期 (YYYYMMDD)",
value="20200101",
className="mb-2",
),
dbc.Input(
id="backtest-end-date",
type="text",
placeholder="结束日期 (YYYYMMDD)",
value="20241231",
className="mb-2",
),
dbc.Button(
"开始回测",
id="start-backtest-btn",
color="success",
size="lg",
className="w-100",
),
]),
],
className="h-100",
),
],
),
# 主内容区
dbc.Col(
width=9,
children=[
# 选股结果
dbc.Card(
[
dbc.CardHeader("选股结果"),
dbc.CardBody([
html.Div(id="stock-select-results"),
]),
],
className="mb-4",
),
# 策略表现
dbc.Card(
[
dbc.CardHeader("策略表现"),
dbc.CardBody([
html.Div(id="strategy-performance"),
]),
],
className="mb-4",
),
# 回测结果
dbc.Card(
[
dbc.CardHeader("回测结果"),
dbc.CardBody([
html.Div(id="backtest-results"),
]),
],
className="mb-4",
),
# 图表
dbc.Card(
[
dbc.CardHeader("图表分析"),
dbc.CardBody([
dcc.Graph(id="stock-chart"),
]),
],
className="mb-4",
),
],
),
]),
],
)
# 初始化组件
selector_engine = SelectorEngine()
backtest_engine = BacktestEngine()
strategy_manager = StrategyManager()
data_manager = DataManager()
# 回调函数：加载策略列表
@app.callback(
Output("strategy-selector", "options"),
Input("strategy-selector", "id")
)
def load_strategy_options(_):
"""加载策略列表到下拉菜单"""
strategies = strategy_manager.list_strategies()
options = [
{"label": f"{strategy['name']} ({strategy['description']})", "value": strategy['name']}
for strategy in strategies
]
return options
# 回调函数：开始选股
@app.callback(
Output("stock-select-results", "children"),
Input("start-select-btn", "n_clicks"),
State("strategy-selector", "value"),
State("adjust-method-selector", "value"),
State("top-n-selector", "value"),
prevent_initial_call=True
)
def start_stock_selection(n_clicks, strategy_name, adjust_method, top_n):
"""开始选股"""
if n_clicks is None:
return html.Div("请点击'开始选股'按钮")
logger.info(f"开始选股：策略={strategy_name}，复权方式={adjust_method}，选股数量={top_n}")
try:
# 执行选股
results = selector_engine.select_top_n(
strategy_name=strategy_name,
n=top_n,
adjust_method=adjust_method
)
if not results:
return dbc.Alert("选股失败，请检查数据和策略配置", color="danger")
# 转换结果为 DataFrame
df = pd.DataFrame(results)
# 显示选股结果
table = dash_table.DataTable(
data=df.to_dict("records"),
columns=[
{"name": "股票代码", "id": "code"},
{"name": "交易日", "id": "trade_date"},
{"name": "收盘价", "id": "close"},
{"name": "成交量", "id": "vol"},
{"name": "得分", "id": "score"}
],
style_table={"overflowX": "auto"},
style_cell={"textAlign": "center"},
style_header={"fontWeight": "bold"},
sort_action="native",
sort_mode="single",
filter_action="native",
page_action="native",
page_current=0,
page_size=10,
)
return table
except Exception as e:
logger.error(f"选股失败：{e}")
return dbc.Alert(f"选股失败：{str(e)}", color="danger")
# 回调函数：更新数据
@app.callback(
Output("strategy-performance", "children"),
Input("update-data-btn", "n_clicks"),
prevent_initial_call=True
)
def update_stock_data(n_clicks):
"""更新股票数据"""
if n_clicks is None:
return html.Div("")
logger.info("开始更新股票数据")
try:
# 更新数据
data_manager.update_all_stocks()
# 获取数据覆盖情况
coverage = selector_engine.get_data_coverage()
coverage_info = dbc.Card(
[
dbc.CardHeader("数据覆盖情况"),
dbc.CardBody([
html.P(f"股票总数: {coverage['total_stocks']}"),
html.P(f"有效数据: {coverage['has_data_count']}"),
html.P(f"无效数据: {coverage['no_data_count']}"),
html.P(f"覆盖率: {coverage['coverage_rate']}%"),
]),
]
)
return coverage_info
except Exception as e:
logger.error(f"数据更新失败：{e}")
return dbc.Alert(f"数据更新失败：{str(e)}", color="danger")
# 回调函数：开始回测
@app.callback(
Output("backtest-results", "children"),
Input("start-backtest-btn", "n_clicks"),
State("strategy-selector", "value"),
State("adjust-method-selector", "value"),
State("backtest-start-date", "value"),
State("backtest-end-date", "value"),
prevent_initial_call=True
)
def start_backtest(n_clicks, strategy_name, adjust_method, start_date, end_date):
"""开始回测"""
if n_clicks is None:
return html.Div("")
logger.info(f"开始回测：策略={strategy_name}，复权方式={adjust_method}，日期={start_date}-{end_date}")
try:
# 执行回测
results = backtest_engine.backtest_strategy(
strategy_name=strategy_name,
adjust_method=adjust_method,
start_date=start_date,
end_date=end_date
)
if not results:
return dbc.Alert("回测失败，请检查参数配置", color="danger")
# 显示回测结果
result_info = dbc.Card(
[
dbc.CardHeader("回测结果"),
dbc.CardBody([
html.P(f"总收益率: {results['total_return']}%"),
html.P(f"年化收益率: {results['annual_return']}%"),
html.P(f"最大回撤: {results['max_drawdown']}%"),
html.P(f"年化波动率: {results['annual_volatility']}%"),
html.P(f"夏普比率: {results['sharpe_ratio']}"),
html.P(f"胜率: {results['win_rate']}%"),
html.P(f"盈亏比: {results['profit_loss_ratio']}"),
html.P(f"交易天数: {results['total_trading_days']}"),
]),
]
)
# 绘制净值曲线
nav_data = pd.DataFrame({
'日期': results['date_series'],
'净值': results['nav_series']
})
nav_chart = dcc.Graph(
id="nav-chart",
figure={
"data": [
{"x": nav_data["日期"], "y": nav_data["净值"], "type": "line", "name": "策略净值"}
],
"layout": {
"title": "策略净值曲线",
"xaxis": {"title": "日期"},
"yaxis": {"title": "净值"},
"height": 300
}
},
)
return html.Div([result_info, nav_chart])
except Exception as e:
logger.error(f"回测失败：{e}")
return dbc.Alert(f"回测失败：{str(e)}", color="danger")
# 回调函数：显示股票图表
@app.callback(
Output("stock-chart", "figure"),
Input("stock-select-results", "children"),
State("adjust-method-selector", "value"),
prevent_initial_call=True
)
def display_stock_chart(children, adjust_method):
"""显示选中股票的图表"""
if not children or "dash_table" not in str(children):
return {
"data": [],
"layout": {
"title": "请先执行选股以查看股票图表",
"xaxis": {"title": "日期"},
"yaxis": {"title": "价格"},
"height": 400
}
}
# 简单返回一个示例图表
return {
"data": [
{
"x": ["2024-01-01", "2024-01-02", "2024-01-03"],
"y": [100, 105, 102],
"type": "line",
"name": "示例股票"
}
],
"layout": {
"title": "股票价格走势（示例数据）",
"xaxis": {"title": "日期"},
"yaxis": {"title": "价格"},
"height": 400
}
}
# 运行应用
def run_app(host: str = None, port: int = None, debug: bool = None):
"""运行 Web 应用"""
if host is None:
host = config.web.host
if port is None:
port = config.web.port
if debug is None:
debug = config.web.debug
logger.info(f"启动 QTrading Web 应用: {host}:{port}")
app.run_server(host=host, port=port, debug=debug)
if __name__ == "__main__":
run_app()