""" Flet UI v2 — 重构版。 分层架构： DataStore — 纯数据，无 UI 依赖 GridPanel — 网格策略持仓表格 + 选中 + 操作栏 DrawerPanel — 右侧抽屉叠加层 + 四 Tab TradeDialogs— 启动/暂停/网格配置弹窗 QmtApp — 生命周期协调 """ import asyncio import time import threading import flet as ft from core.qmt_real import RealQmtV, qmtv from core.logger import LogLevel, PrintLog from core.sfgrid.model import SFGridTradeTarget, STRATEGY_TYPE_GRID from core.sfgrid.sfgrid_strategy import SFGridStrategy from core.eventbus import event_bus, MarketDataUpdate, EventMarketActiveSwitch from core.sfgrid.bus_events import EventTradeTargetUpdate, EventPoolMark # ── 工具函数 ── _ORDER_STATUS = {48: '未报', 49: '待报', 50: '已报', 51: '已报待撤', 52: '部成待撤', 53: '部撤', 54: '已撤', 55: '部成', 56: '已成', 57: '废单'} def _fmt_time(t) -> str: if not t: return '' import datetime try: ts = int(t) if ts > 1e12: ts //= 1000 return datetime.datetime.fromtimestamp(ts).strftime('%H:%M:%S') except (ValueError, OSError): return str(t) def _direction(ot: int) -> str: return '买' if ot == 23 else '卖' if ot == 24 else str(ot) def _plain(code: str) -> str: return code.split('.')[0] if '.' in code else code def _text(s, color=None, bold=False, size=None) -> ft.Text: """快捷构建 Text""" return ft.Text(str(s), color=color, weight=ft.FontWeight.BOLD if bold else None, size=size) def _datatable(cols: list[str], rows: list[list[str]], col_widths: list = None) -> ft.DataTable: """构建 DataTable（不包装 ListView）""" data_rows = [] for r in rows: cells = [] for i, c in enumerate(r): w = (col_widths or [None] * len(r))[i] if i < len(r) else None cells.append(ft.DataCell(ft.Text(str(c), overflow=ft.TextOverflow.ELLIPSIS, max_lines=1, width=w))) data_rows.append(ft.DataRow(cells=cells)) if not data_rows: data_rows.append(ft.DataRow(cells=[ft.DataCell(ft.Text("")) for _ in cols])) return ft.DataTable(columns=[ft.DataColumn(ft.Text(h)) for h in cols], rows=data_rows, width=float('inf'), heading_row_height=36, data_row_min_height=32) # ══════════════════════════════════════════════════════════════ # DataStore — 纯数据层 # ══════════════════════════════════════════════════════════════ class _DataStore: """持仓、行情、订单数据管理，不依赖任何 UI""" def __init__(self): self.tradeTargets: dict[int, SFGridTradeTarget] = {} self.stockCodeIdMap: dict[str, int] = {} self.marketPrices: dict[int, float] = {} # tid → lastPrice self.preClose: dict[int, float] = {} # tid → 昨收 self.avgPrices: dict[int, float] = {} # tid → avgPrice self.orders: list = [] self.trades: list = [] self.strategyCtrl: dict[int, SFGridStrategy] = {} # 市场监控 self.monitorPrice: float = 10.0 self.marketLog: dict[str, dict] = {} # stock_code → {name, price, time} self.listeningStocks: list = [] self.poolLog: list[str] = [] # 股票池管理器日志（list，方便 append） # ── 初始化 ── def load_from_qmt(self): """从 QMT 加载持仓并导入 DB""" positions = qmtv.getAllPositions() PrintLog(LogLevel.INFO, f'[Data] 持仓: {len(positions)} 个') for code, pos in positions.items(): existing = SFGridTradeTarget.get_or_none(SFGridTradeTarget.stock_code == code) if existing is None: name = getattr(pos, 'instrument_name', '') or qmtv.getInstrumentName(code) SFGridTradeTarget.create( stock_code=code, stock_name=name, current_position=int(pos.volume), init_price=float(getattr(pos, 'avg_price', 0) or 0), grid_index=0, enabled=False, grid_start_price=float(getattr(pos, 'avg_price', 0) or 0) or 10.0, grid_size=1.0, grid_volume=200, grid_upper_count=1, grid_lower_count=10, ) # 映射 (先建立 stockCodeIdMap) results = list(SFGridTradeTarget.select()) for t in results: pos = positions.get(t.stock_code) t.current_position = 0 if pos is None else int(pos.volume) tid = t.get_id() self.tradeTargets[tid] = t self.stockCodeIdMap[t.stock_code] = tid if pos is not None: self.avgPrices[tid] = float(getattr(pos, 'avg_price', 0) or 0) # 昨收（依赖 stockCodeIdMap） self._load_preclose(positions) def _load_preclose(self, positions: dict): try: from xtquant import xtdata for stock_code in positions: if '.' not in stock_code: full_code = f'{stock_code}.SH' if stock_code.startswith(('6','5','9')) else f'{stock_code}.SZ' else: full_code = stock_code detail = xtdata.get_instrument_detail(full_code) if detail: pc = detail.get('PreClose', 0) if isinstance(detail, dict) else getattr(detail, 'PreClose', 0) if pc > 0 and stock_code in self.stockCodeIdMap: self.preClose[self.stockCodeIdMap[stock_code]] = float(pc) PrintLog(LogLevel.INFO, f'[Data] 昨收: {len(self.preClose)} 个') except Exception as e: PrintLog(LogLevel.DEBUG, f'[Data] 昨收异常: {e}') def init_strategies(self): for tid, t in self.tradeTargets.items(): if t.strategy_type == STRATEGY_TYPE_GRID and t.enabled: self.strategyCtrl[tid] = SFGridStrategy(t) def remove_target(self, tid: int) -> str: """移除/降级网格策略标的。返回 'unclassified'（有持仓降级）或 'deleted'（无持仓删除）""" target = self.tradeTargets.get(tid) if target is None: return 'none' # 先停止策略 ctrl = self.strategyCtrl.pop(tid, None) if ctrl: ctrl.enabledTrading(False) if target.current_position > 0: # 有持仓 → 降级为未分类 target.strategy_type = 0 # STRATEGY_TYPE_UNCLASSIFIED target.enabled = False target.save() return 'unclassified' else: # 无持仓 → 直接删除 target.delete_instance() self.tradeTargets.pop(tid, None) self.stockCodeIdMap.pop(target.stock_code, None) self.marketPrices.pop(tid, None) self.avgPrices.pop(tid, None) self.preClose.pop(tid, None) return 'deleted' # ── 行情 ── def apply_tick(self, data: dict) -> int: """应用行情数据，返回更新的标的数""" count = 0 for stock_code, tick in data.items(): plain = _plain(stock_code) tid = self.stockCodeIdMap.get(plain) lp = tick.get('lastPrice', 0) if tid is not None and tid in self.tradeTargets: self.marketPrices[tid] = lp self.tradeTargets[tid].market_price = lp count += 1 elif lp == self.monitorPrice or stock_code in self.listeningStocks: if stock_code not in self.listeningStocks: self.listeningStocks.append(stock_code) self.marketLog[stock_code] = { 'stock_name': qmtv.getInstrumentName(stock_code), 'last_price': lp, 'time': time.strftime("%H:%M:%S"), } return count def pull_prices(self): """主动拉取缺失的市价""" for tid, t in self.tradeTargets.items(): if tid not in self.marketPrices or self.marketPrices[tid] == 0: price = qmtv.getLastPrice(t.stock_code) if price > 0: self.marketPrices[tid] = price t.market_price = price # ── 持仓 ── def refresh_positions(self): positions = qmtv.getAllPositions() for t in self.tradeTargets.values(): pos = positions.get(t.stock_code) t.current_position = 0 if pos is None else int(pos.volume) # ── 委托 / 成交 ── def refresh_orders(self): try: self.orders = list(qmtv.queryTodayOrders()) except Exception: pass def refresh_trades(self): try: self.trades = list(qmtv.queryTodayTrades()) except Exception: pass def active_orders(self) -> list: """返回过滤+去重后的活跃委托行""" # 过滤已终结的订单: 已撤(54) 已成(56) 废单(57) _TERMINAL = {54, 56, 57} o_map = {} for o in self.orders: oid = str(getattr(o, 'order_id', '')) if not oid: continue o_map[oid] = o rows = [] for o in o_map.values(): st = getattr(o, 'order_status', 0) if st in _TERMINAL: continue tv = getattr(o, 'traded_volume', 0) or 0 ov = getattr(o, 'order_volume', 0) or 0 rows.append([ _fmt_time(getattr(o, 'order_time', 0)), _plain(getattr(o, 'stock_code', '')), getattr(o, 'instrument_name', '') or '', _direction(getattr(o, 'order_type', 0)), f"{getattr(o, 'price', 0):.3f}", f"{tv}/{ov}", f"{getattr(o, 'traded_price', 0):.3f}" if getattr(o, 'traded_price', 0) > 0 else '-', _ORDER_STATUS.get(st, '未知'), ]) return rows def active_trades(self) -> list: """返回去重后的成交行""" t_map = {} for t in self.trades: tid = str(getattr(t, 'traded_id', '')) if not tid: continue t_map[tid] = t rows = [] for t in t_map.values(): rows.append([ _fmt_time(getattr(t, 'traded_time', 0)), _plain(getattr(t, 'stock_code', '')), getattr(t, 'instrument_name', '') or '', _direction(getattr(t, 'order_type', 0)), f"{getattr(t, 'traded_price', 0):.3f}", str(getattr(t, 'traded_volume', 0)), f"{getattr(t, 'traded_amount', 0):.2f}", f"{getattr(t, 'commission', 0):.2f}", ]) return rows def add_from_market(self, stock_code: str, stock_name: str) -> int: """从市场监控添加标的（默认: grid_index=0, upper=0, lower=10），返回 tid""" plain = _plain(stock_code) existing = SFGridTradeTarget.get_or_none(SFGridTradeTarget.stock_code == plain) if existing: tid = existing.get_id() if tid not in self.tradeTargets: self.tradeTargets[tid] = existing self.stockCodeIdMap[plain] = tid return tid mp = self.marketLog.get(stock_code, {}) price = mp.get('last_price', 10.0) target = SFGridTradeTarget.create( stock_code=plain, stock_name=stock_name, current_position=0, init_price=0.0, grid_index=0, enabled=False, strategy_type=STRATEGY_TYPE_GRID, grid_start_price=float(price) if price > 0 else 10.0, grid_size=1.0, grid_volume=200, grid_upper_count=0, grid_lower_count=10, ) tid = target.get_id() self.tradeTargets[tid] = target self.stockCodeIdMap[plain] = tid # 从市场监控列表中移除（加入交易池后不再监控） self.marketLog.pop(stock_code, None) if stock_code in self.listeningStocks: self.listeningStocks.remove(stock_code) return tid def pending_tags(self, stock_code: str) -> list: """某标的挂单方向标签""" tags = [] for o in self.orders: if _plain(getattr(o, 'stock_code', '')) != stock_code: continue if getattr(o, 'order_status', 0) in {54, 57}: continue ot = getattr(o, 'order_type', 0) if ot == 23 and '多' not in tags: tags.append('多') elif ot == 24 and '空' not in tags: tags.append('空') return tags def pool_action_tags(self, stock_code: str) -> list[str]: """某标的的股票池待执行标记（eliminate/liquidate）""" plain = _plain(stock_code) from core.scoring.models import PendingPoolAction rows = list(PendingPoolAction .select(PendingPoolAction.action_type) .where(PendingPoolAction.stock_code == plain) .dicts()) tags = [] for r in rows: if r['action_type'] == 'eliminate' and '淘汰' not in tags: tags.append('淘汰') elif r['action_type'] == 'liquidate' and '沉寂' not in tags: tags.append('沉寂') return tags # ══════════════════════════════════════════════════════════════ # GridPanel — 网格策略持仓表格 # ══════════════════════════════════════════════════════════════ class _GridPanel: """网格策略持仓面板 — 自定义 Row 表格，行内按钮直接可用""" def __init__(self, data: _DataStore, dialogs=None): self._data = data self._dialogs = dialogs # _TradeDialogs self._col = None @property def controls(self): return self._col.controls if self._col else [] def build(self) -> ft.Control: self._col = ft.Column(spacing=0, scroll=ft.ScrollMode.AUTO, expand=True) self._rebuild() return self._col def _rebuild(self): # (width, expand): 0=固定宽, >0=弹性比重; 新增排名列(50px) _C = [(35, 0), (0, 2), (70, 1), (80, 1), (50, 1), (55, 1), (50, 0), (60, 1), (0, 1)] H = ["ID", "股票", "市场价", "网格基准", "持仓", "成本", "排名", "状态", "操作"] header = [] for h, (w, e) in zip(H, _C): if e > 0: header.append(ft.Container(_text(h, bold=True), padding=4, expand=e)) else: header.append(ft.Container(_text(h, bold=True), width=w, padding=4)) rows = [ft.Row(header, spacing=0), ft.Divider(height=1, color='#e0e0e0')] # 获取最新评分排名 rank_map = {} # stock_code -> score_rank try: from core.scoring.models import ScoringResult from peewee import fn latest_date = ScoringResult.select(fn.MAX(ScoringResult.trade_date)).scalar() if latest_date: scored = (ScoringResult .select(ScoringResult.stock_code, ScoringResult.score_rank) .where(ScoringResult.trade_date == latest_date) .dicts()) for r in scored: rank_map[r['stock_code']] = r['score_rank'] except Exception: pass for tid, t in self._data.tradeTargets.items(): if t.strategy_type != STRATEGY_TYPE_GRID: continue pg = t.getPriceGrid() idx = t.grid_index base = pg[idx] if 0 <= idx < len(pg) else 0 mp = self._data.marketPrices.get(tid, 0) or 0 pc = self._data.preClose.get(tid, 0) or 0 pcolor = '#CC0000' if mp > pc > 0 else '#009900' if mp < pc and mp > 0 else None # 网格基准列 gcells = [_text(f'{base:.2f}', bold=True, )] if mp > base > 0: gcells.append(_text(' ▲', color='#CC0000', bold=True, )) elif 0 < mp < base: gcells.append(_text(' ▼', color='#009900', bold=True, )) for tag in self._data.pending_tags(t.stock_code): gcells.append(ft.Container( _text(tag, color='white', bold=True, size=10), bgcolor='#E67E22' if tag == '多' else '#3498DB', border_radius=4, padding=ft.Padding(3, 1, 3, 1), )) gcell = ft.Row(gcells, spacing=3) if len(gcells) > 1 else gcells[0] # 排名列 plain = t.stock_code.split('.')[0] if '.' in t.stock_code else t.stock_code rank = rank_map.get(plain, None) if rank is not None and rank <= 50: rank_str = f'#{rank}' rank_color = '#4CAF50' elif rank is not None and rank <= 100: rank_str = f'#{rank}' rank_color = '#2196F3' elif rank is not None: rank_str = f'#{rank}' rank_color = '#888888' else: rank_str = '—' rank_color = '#888888' # 操作按钮 _ICON = 22 # 图标大小，比默认 18 好按 if t.enabled: btns = [ ft.IconButton(ft.Icons.PAUSE, icon_size=_ICON, tooltip="暂停", on_click=lambda e, tt=t: self._dialogs.confirm_stop(tt)), ft.IconButton(ft.Icons.SETTINGS, icon_size=_ICON, tooltip="设置", icon_color='#BDBDBD', on_click=lambda e, tt=t: self._dialogs.open_config(tt)), ft.IconButton(ft.Icons.REMOVE, icon_size=_ICON, tooltip="运行中不可移除", icon_color='#BDBDBD', on_click=None), ] else: btns = [ ft.IconButton(ft.Icons.PLAY_ARROW, icon_size=_ICON, tooltip="启动", on_click=lambda e, tt=t: self._dialogs.confirm_start(tt)), ft.IconButton(ft.Icons.SETTINGS, icon_size=_ICON, tooltip="设置", on_click=lambda e, tt=t: self._dialogs.open_config(tt)), ft.IconButton(ft.Icons.REMOVE, icon_size=_ICON, tooltip="移除", on_click=lambda e, tt=t: self._dialogs.confirm_remove(tt)), ] # 股票列：名称 + 标记标签 pool_tags = self._data.pool_action_tags(t.stock_code) tag_chips = [] if pool_tags: # 有淘汰/沉寂时，显示操作标签，不显示默认标签 for tag in pool_tags: color = '#E67E22' if tag == '淘汰' else '#E74C3C' tag_chips.append(ft.Container( _text(tag, color='white', bold=True, size=10), bgcolor=color, border_radius=4, padding=ft.Padding(2, 1, 2, 1), )) else: # 无操作标签时，显示默认"网格"标签 tag_chips.append(ft.Container( _text('网格', color='white', bold=True, size=10), bgcolor='#2E7D32', border_radius=4, padding=ft.Padding(2, 1, 2, 1), )) name_cell = ft.Row([_text(f'{t.stock_code} {t.stock_name}')] + tag_chips, spacing=4) cells_text = [str(tid), name_cell, f'{mp:.3f}', gcell, str(t.current_position), f'{self._data.avgPrices.get(tid, 0):.3f}', rank_str, '▶运行中' if t.enabled else '⏸已暂停', btns] row_cells = [] for i, (w, e) in enumerate(_C): content = cells_text[i] if i == 2: # 市场价用颜色 content = _text(cells_text[i], color=pcolor, bold=True) elif i == 6: # 排名用颜色 content = _text(cells_text[i], color=rank_color, bold=True) elif isinstance(content, str): content = _text(content) elif isinstance(content, list): content = ft.Row(content, spacing=6) if e > 0: row_cells.append(ft.Container(content, padding=4, expand=e)) else: row_cells.append(ft.Container(content, width=w, padding=4)) row = ft.Row(row_cells, spacing=0) rows.append(ft.Container(row, padding=ft.Padding(0, 2, 0, 2))) rows.append(ft.Divider(height=1, color='#f0f0f0')) self._col.controls = rows # ══════════════════════════════════════════════════════════════ # DrawerPanel — 右侧抽屉面板 # ══════════════════════════════════════════════════════════════ class _DrawerPanel: """右侧叠加抽屉：市场监控 + 委托 + 成交 + 未分类""" def __init__(self, data: _DataStore, dialogs=None, page=None): self._data = data self._dialogs = dialogs self._tab_orders = ft.Tab(label="当前委托") self._tab_trades = ft.Tab(label="当日成交") self._tab_dataset = ft.Tab(label="数据集管理") self._tab_scoring = ft.Tab(label="每日评分") self._order_table = None self._trade_table = None self._uncl_col = None self._market_table = None self._dataset_table = None self._scoring_table = None self._sync_lock = False # 同步线程锁 self._sync_btns = [] # 同步按钮引用列表 def build(self) -> ft.Control: """返回 overlay Container""" self._uncl_col = ft.Column(spacing=0, scroll=ft.ScrollMode.AUTO, expand=True) self._order_table = ft.ListView(expand=True) self._trade_table = ft.ListView(expand=True) self._market_table = ft.ListView(expand=True) self._dataset_table = ft.ListView(expand=True) self._dataset_col = ft.Column(scroll=ft.ScrollMode.AUTO, expand=True) self._scoring_table = ft.ListView(expand=True) self._backtest_col = ft.Column(scroll=ft.ScrollMode.AUTO, expand=True) bar = ft.TabBar(tabs=[ ft.Tab(label="实时价格监控"), self._tab_scoring, self._tab_orders, self._tab_trades, self._tab_dataset, ft.Tab(label="未分类持仓"), ]) view = ft.TabBarView(controls=[ self._build_market_tab(), self._build_scoring_tab(), self._order_table, self._trade_table, self._build_dataset_tab(), self._uncl_col, ], expand=True) panel = ft.Container( ft.Column([ ft.Tabs(ft.Column([bar, view], expand=True), length=6, expand=True), ], expand=True), width=750, bgcolor=ft.Colors.SURFACE, ) backdrop = ft.Container(bgcolor='#44000000', expand=True) return ft.Container(ft.Row([backdrop, panel], spacing=0), visible=False, expand=True) def _build_market_tab(self) -> ft.Control: price_input = ft.TextField(value=str(self._data.monitorPrice), width=80, height=32, text_size=13) return ft.Column([ ft.Row([_text("监控配置", size=13), _text("价格", size=13), price_input, ft.ElevatedButton("确认", on_click=lambda e: self._set_monitor(price_input.value), height=32)]), ft.Container(content=self._market_table, expand=True), ], expand=True) def _set_monitor(self, val: str): try: self._data.monitorPrice = float(val) self._data.marketLog.clear() self._data.listeningStocks.clear() except ValueError: pass # ── 刷新 ── def refresh_all(self): self._refresh_grid() self._refresh_orders() self._refresh_trades() self._refresh_market() self._refresh_dataset() self._refresh_scoring() def _refresh_grid(self): # (width, expand): 0=固定宽, >0=弹性比重; 股票列 expand 自动填充剩余空间 # 新增: 排名列 (50px) _C = [(35, 0), (0, 1), (80, 0), (60, 0), (70, 0), (50, 0), (65, 0)] H = ["ID", "股票", "市场价", "持仓", "成本", "排名", "操作"] # 获取最新评分排名 rank_map = {} # stock_code -> score_rank try: from core.scoring.models import ScoringResult from peewee import fn latest_date = ScoringResult.select(fn.MAX(ScoringResult.trade_date)).scalar() if latest_date: rows = (ScoringResult .select(ScoringResult.stock_code, ScoringResult.score_rank) .where(ScoringResult.trade_date == latest_date) .dicts()) for r in rows: rank_map[r['stock_code']] = r['score_rank'] except Exception: pass # 数据库未就绪时忽略 def _cell(text, w, e, color=None): if e > 0: return ft.Container(_text(text, color=color), padding=4, expand=e) return ft.Container(_text(text, color=color), width=w, padding=4) header = ft.Row([_cell(h, w, e) for h, (w, e) in zip(H, _C)], spacing=0) rows = [header, ft.Divider(height=1, color='#e0e0e0')] for tid, t in self._data.tradeTargets.items(): if t.strategy_type == STRATEGY_TYPE_GRID: continue mp = self._data.marketPrices.get(tid, 0) or 0 plain = t.stock_code.split('.')[0] if '.' in t.stock_code else t.stock_code rank = rank_map.get(plain, None) rank_str = f"#{rank}" if rank is not None else "—" # 排名颜色: top50 绿色, top100 蓝色, 其他灰色 if rank is not None and rank <= 50: rank_color = '#4CAF50' elif rank is not None and rank <= 100: rank_color = '#2196F3' else: rank_color = '#888888' cells = [ _cell(str(tid), *_C[0]), _cell(f'{t.stock_code} {t.stock_name}', *_C[1]), _cell(f'{mp:.3f}', *_C[2]), _cell(str(t.current_position), *_C[3]), _cell(f'{self._data.avgPrices.get(tid, 0):.3f}', *_C[4]), _cell(rank_str, *_C[5], color=rank_color), ft.Container(ft.IconButton(ft.Icons.SETTINGS, icon_size=20, tooltip="网格配置", on_click=lambda e, tt=t: self._dialogs.open_config(tt)), width=_C[6][0], padding=0), ] row = ft.Row(cells, spacing=0) rows.append(ft.Container(row, padding=ft.Padding(0, 2, 0, 2))) rows.append(ft.Divider(height=1, color='#f0f0f0')) self._uncl_col.controls = rows def _refresh_orders(self): rows = self._data.active_orders() self._tab_orders.label = f"当前委托 ({len(rows)})" if rows else "当前委托" cols = ["时间", "代码", "名称", "方向", "委托价", "已成交/委托量", "均价", "状态"] self._order_table.controls = [ft.ListView( [_datatable(cols, rows, [65,55,70,35,60,80,55,50])], expand=True)] def _refresh_trades(self): rows = self._data.active_trades() self._tab_trades.label = f"当日成交 ({len(rows)})" if rows else "当日成交" cols = ["时间", "代码", "名称", "方向", "成交价", "成交量", "成交金额", "手续费"] self._trade_table.controls = [ft.ListView( [_datatable(cols, rows, [65,55,70,35,65,60,70,55])], expand=True)] def _refresh_market(self): # (width, expand): 0=固定宽, >0=弹性比重; 股票名称列 expand 自动填充剩余空间 _C = [(80, 0), (0, 1), (80, 0), (45, 0)] H = ["时间", "股票名称", "最新价格", "操作"] def _header_cell(text, w, e): if e > 0: return ft.Container(_text(text, bold=True), padding=4, expand=e) return ft.Container(_text(text, bold=True), width=w, padding=4) header = ft.Row([_header_cell(h, w, e) for h, (w, e) in zip(H, _C)], spacing=0) rows = [header, ft.Divider(height=1, color='#e0e0e0')] for sc, d in self._data.marketLog.items(): plain = _plain(sc) already_in_pool = plain in self._data.stockCodeIdMap def _data_cell(text, w, e, color=None, size=None): if e > 0: return ft.Container(_text(text, color=color, size=size), padding=4, expand=e) return ft.Container(_text(text, color=color, size=size), width=w, padding=4) cells = [ _data_cell(d.get('time', ''), *_C[0]), _data_cell(f"{d['stock_name']}-{sc}", *_C[1]), _data_cell(f"{d['last_price']:.3f}", *_C[2]), ] if already_in_pool: op_w, _ = _C[3] cells.append(ft.Container( ft.Row([ft.Text("已在池", color='#AAAAAA', size=11, text_align=ft.TextAlign.CENTER)], alignment=ft.MainAxisAlignment.CENTER), width=op_w, padding=0)) else: cells.append(ft.Container( ft.Row([ft.IconButton(ft.Icons.ADD, icon_size=20, tooltip="添加持仓", on_click=lambda e, code=sc: self._on_add_from_market(code))], alignment=ft.MainAxisAlignment.CENTER), width=_C[3][0], padding=0)) row = ft.Row(cells, spacing=0) rows.append(ft.Container(row, padding=ft.Padding(0, 2, 0, 2))) rows.append(ft.Divider(height=1, color='#f0f0f0')) self._market_table.controls = [ft.Column(rows, spacing=0)] # ── Tab 5: 数据集管理 ── # 按依赖顺序: kline → stocks → industry → market → sector _SYNC_DEPS = { 'kline': ['kline'], 'stocks': ['stocks'], 'industry':['industry'], 'market': ['market'], 'sector': ['sector'], 'all': ['kline', 'stocks', 'industry', 'market', 'sector'], } _SYNC_LABELS = { 'kline': '同步K线', 'stocks': '同步股票信息', 'industry':'同步行业映射', 'market': '计算市场状态', 'sector': '计算行业指数', } def _build_dataset_tab(self) -> ft.Control: self._dataset_status = ft.Text("就绪", size=12, color='#888888') self._backtest_label = ft.Text("v6.7r3 回测曲线", size=11, color='#aaaaaa') self._backtest_img = ft.Container(visible=False) # 回测图表占位，运行时替换为 Image return ft.Column([ self._dataset_status, self._dataset_col, ft.Divider(height=2, color='#444444'), self._backtest_label, self._backtest_img, ], expand=True, spacing=6, scroll=ft.ScrollMode.AUTO) def _run_sync(self, targets: list): # 线程锁 if self._sync_lock: self._dataset_status.value = "同步中，跳过重复点击" self._dataset_status.update() return self._sync_lock = True self._set_sync_btns_disabled(True) self._dataset_status.value = "同步中..." self._dataset_status.update() import threading def _do(): from core.scoring.sync import ( KlineStockSync, KlineIndexSync, StocksSync, IndustrySync, MarketRegimeSync, SectorFeaturesSync, ) syncs = { 'kline': [KlineStockSync(count=300), KlineIndexSync(count=300)], 'stocks': [StocksSync()], 'industry':[IndustrySync()], 'market': [MarketRegimeSync()], 'sector': [SectorFeaturesSync()], } order = [] for t in targets: for key in self._SYNC_DEPS.get(t, [t]): order.extend(syncs.get(key, [])) for s in order: try: s.run() except Exception: pass self._sync_lock = False self._set_sync_btns_disabled(False) self._dataset_status.value = "同步完成" self._dataset_status.update() self._refresh_dataset() threading.Thread(target=_do, daemon=True).start() def _refresh_backtest(self): """加载 v6.7r3 回测图表，嵌入数据集 Tab""" import pandas as pd import io, base64, json import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt import matplotlib.dates as mdates from pathlib import Path model_dir = Path(__file__).parent.parent.parent.parent / 'models' daily_csv = model_dir / 'backtest_v67r3_daily.csv' summary_json = model_dir / 'backtest_v67r3_summary.json' if not daily_csv.exists(): self._backtest_label.value = "回测数据不存在" self._backtest_label.update() return try: df_daily = pd.read_csv(daily_csv, parse_dates=['date']) summary = json.loads(summary_json.read_text()) if summary_json.exists() else {} # 更新标题行 start_val = summary.get('initial_cash', 60000) final_val = summary.get('final_total_value', 0) self._backtest_label.value = (f"v6.7r3 回测 初始:{start_val:,.0f} → 终值:{final_val:,.0f} " f"收益:{summary.get('total_return_pct',0):.1f}% " f"夏普:{summary.get('annual_sharpe',0):.2f} " f"回撤:{summary.get('max_drawdown_pct',0):.1f}%") self._backtest_label.update() # 渲染总资产曲线 import matplotlib.font_manager as mfont # 查找中文字体 cjk_names = ['Microsoft YaHei', 'SimHei', 'Noto Sans SC', 'WenQuanYi'] cjk_font = next((f.fname for f in mfont.fontManager.ttflist if any(n in f.name for n in cjk_names)), None) if cjk_font: prop = mfont.FontProperties(fname=cjk_font) plt.rcParams['font.family'] = prop.get_name() plt.rcParams['axes.unicode_minus'] = False fig, ax = plt.subplots(figsize=(7, 3.5), dpi=100) fig.patch.set_facecolor('#1e1e1e') ax.set_facecolor('#2d2d2d') ax.plot(df_daily['date'], df_daily['total_asset'], color='#4CAF50', linewidth=1.5) ax.fill_between(df_daily['date'], df_daily['total_asset'], alpha=0.1, color='#4CAF50') ax.set_title('总资产曲线 (v6.7r3)', color='#ffffff', fontsize=10) ax.set_ylabel('元', color='#cccccc', fontsize=9) ax.tick_params(colors='#cccccc', labelsize=8) ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m')) ax.xaxis.set_major_locator(mdates.MonthLocator(interval=3)) plt.setp(ax.xaxis.get_majorticklabels(), rotation=30, ha='right') for spine in ax.spines.values(): spine.set_color('#555555') ax.grid(True, alpha=0.15, color='#888888') # 标注起点终点 ax.annotate(f'{start_val:,.0f}', xy=(df_daily['date'].iloc[0], start_val), xytext=(3, 5), textcoords='offset points', color='#aaaaaa', fontsize=8) ax.annotate(f'{final_val:,.0f}', xy=(df_daily['date'].iloc[-1], final_val), xytext=(3, 5), textcoords='offset points', color='#4CAF50', fontsize=8, fontweight='bold') plt.tight_layout() buf = io.BytesIO() fig.savefig(buf, format='png', bbox_inches='tight', facecolor=fig.get_facecolor()) buf.seek(0) img_b64 = base64.b64encode(buf.read()).decode() plt.close(fig) # 用实际 Image 替换占位的 Container new_img = ft.Image( src=f'data:image/png;base64,{img_b64}', fit=ft.BoxFit.CONTAIN, height=240, visible=True, ) # 找到 placeholder Container 在 Column 中的位置，替换 idx = self._backtest_label.parent.controls.index(self._backtest_img) self._backtest_label.parent.controls[idx] = new_img self._backtest_img = new_img self._backtest_label.parent.update() except Exception as ex: self._backtest_label.value = f"图表加载失败: {ex}" self._backtest_label.color = '#F44336' self._backtest_label.update() def _set_sync_btns_disabled(self, disabled: bool): for btn in self._sync_btns: btn.disabled = disabled btn.update() def _refresh_dataset(self): from core.scoring.models import ( KlineStock, KlineIndex, StockInfo, IndustryMapping, MarketRegimeDaily, SectorFeaturesDaily, ScoringResult, ) tables = [ ("kline_stock", KlineStock, "kline", "trade_date"), ("kline_index", KlineIndex, "kline", "trade_date"), ("stocks", StockInfo, "stocks", "share_updated_at"), ("industry", IndustryMapping, "industry","update_date"), ("market_regime", MarketRegimeDaily, "market", "trade_date"), ("sector_features", SectorFeaturesDaily, "sector", "trade_date"), ("ScoringResult", ScoringResult, "all", "trade_date"), ] # 与市场监控Tab一致的列宽布局: (width, expand) _C = [(150, 0), (80, 0), (100, 0)] H = ["表名", "记录数", "最新日期"] def _hcell(text, w, e): if e > 0: return ft.Container( ft.Text(text, weight=ft.FontWeight.BOLD), padding=4, expand=e) return ft.Container( ft.Text(text, weight=ft.FontWeight.BOLD), width=w, padding=4) header = ft.Row([_hcell(h, w, e) for h, (w, e) in zip(H, _C)], spacing=0) self._sync_btns.clear() if self._sync_lock: for btn in self._sync_btns: btn.disabled = True rows = [header, ft.Divider(height=1, color='#e0e0e0')] for name, model_cls, dep_key, date_col in tables: try: cnt = model_cls.select().count() df = getattr(model_cls, date_col, None) if df is not None: last = (model_cls.select(df).order_by(df.desc()).first()) date_str = str(getattr(last, date_col, "")) if last else "" else: date_str = "" except Exception: cnt, date_str = "ERR", "" def _dcell(text, w, e=0): if e > 0: return ft.Container(ft.Text(text), padding=4, expand=e) return ft.Container(ft.Text(text), width=w, padding=4) cells = [ _dcell(name, *_C[0]), _dcell(str(cnt), *_C[1]), _dcell(date_str, *_C[2]), ] row = ft.Row(cells, spacing=0) rows.append(ft.Container(row, padding=ft.Padding(0, 2, 0, 2))) rows.append(ft.Divider(height=1, color='#f0f0f0')) self._dataset_col.controls = rows self._refresh_backtest() # ── Tab 6: 每日评分 ── def _build_scoring_tab(self) -> ft.Control: from datetime import date, timedelta self._score_cur_date = date.today() self._score_date_label = ft.Text(str(self._score_cur_date), size=14, weight=ft.FontWeight.BOLD) self._score_status = ft.Text("无评分数据", size=12, color='#888888') def _prev_day(_): self._score_cur_date -= timedelta(days=1) self._score_date_label.value = str(self._score_cur_date) self._score_date_label.update() self._refresh_scoring() def _next_day(_): self._score_cur_date += timedelta(days=1) self._score_date_label.value = str(self._score_cur_date) self._score_date_label.update() self._refresh_scoring() def _today(_): self._score_cur_date = date.today() self._score_date_label.value = str(self._score_cur_date) self._score_date_label.update() self._refresh_scoring() date_row = ft.Row([ ft.IconButton(ft.Icons.CHEVRON_LEFT, icon_size=22, tooltip="前一天", on_click=_prev_day), self._score_date_label, ft.IconButton(ft.Icons.CHEVRON_RIGHT, icon_size=22, tooltip="后一天", on_click=_next_day), ft.IconButton(ft.Icons.TODAY, icon_size=20, tooltip="回到今天", on_click=_today), ft.Container(ft.Divider(height=20), width=2), ft.ElevatedButton("同步数据", on_click=lambda e: self._run_sync(['kline', 'stocks', 'industry', 'market', 'sector']), height=32), ft.ElevatedButton("执行评分", on_click=lambda e: self._run_scoring(), height=32), ft.Container(ft.Divider(height=20), width=2), ft.ElevatedButton("沉寂检测", on_click=lambda e: self._data._pool_manager.trigger_slumber(), height=32, tooltip="手动触发沉寂检测（阶段一标记）"), ft.ElevatedButton("淘汰检测", on_click=lambda e: self._data._pool_manager.trigger_weekly_elim(), height=32, tooltip="手动触发周度淘汰检测（阶段一标记）"), self._score_status, ], spacing=6, vertical_alignment=ft.CrossAxisAlignment.CENTER) return ft.Column([ date_row, self._scoring_table, ], expand=True, spacing=6) def _run_scoring(self): self._score_status.value = "评分中..." self._score_status.update() import threading def _do(): from core.scoring.inference.scorer import GridSeekerPipeline trade_date = self._score_cur_date # 检查当日 K线数据是否已同步 from peewee import fn from core.scoring.models import KlineStock latest_kline = KlineStock.select(fn.MAX(KlineStock.trade_date)).scalar() if latest_kline is None or latest_kline < trade_date: self._score_status.value = f"K线未同步至 {trade_date}，请先盘后同步" self._score_status.color = '#F44336' self._score_status.update() return try: engine = GridSeekerPipeline() rankings = engine.run(trade_date) if not rankings.empty: engine.persist(rankings, trade_date) self._score_status.value = f"{trade_date} 评分完成" except FileNotFoundError: self._score_status.value = "模型文件缺失" except Exception as ex: self._score_status.value = f"失败: {ex}" self._score_status.update() self._refresh_scoring() threading.Thread(target=_do, daemon=True).start() def _refresh_scoring(self): from core.scoring.models import ScoringResult from core.qmt import qmtv # (width, expand): 排名60 + 名称弹性 + 概率65 + 轮数58 + 堆叠概率65 + 操作32 _C = [(60, 0), (0, 1), (65, 1), (58, 1), (65, 1), (32, 1)] H = ["排名", "代码 / 名称", "Rank轮数", "Top概率", "Stacking概率", "操作"] def _hcell(text, w, e): if e > 0: return ft.Container(_text(text, bold=True), padding=4, expand=e) return ft.Container(_text(text, bold=True), width=w, padding=4) def _dcell(text, w, e, color=None, size=None): if e > 0: return ft.Container(_text(text, color=color, size=size), padding=4, expand=e) return ft.Container(_text(text, color=color, size=size), width=w, padding=4) header = ft.Row([_hcell(h, w, e) for h, (w, e) in zip(H, _C)], spacing=0) rows = [header, ft.Divider(height=1, color='#e0e0e0')] # 用导航日期查询 trade_date = self._score_cur_date scored_rows = list((ScoringResult .select() .where(ScoringResult.trade_date == trade_date) .order_by(ScoringResult.score_rank) .limit(200) .dicts())) if not scored_rows: self._score_status.value = f"{trade_date} 无评分数据" self._score_status.color = '#F44336' self._scoring_table.controls = rows self._score_status.update() return shown = 0 for r in scored_rows: code = r['stock_code'] plain = code.split('.')[0] if '.' in code else code # 过滤 ST from core.scoring.models import StockInfo if plain.startswith(('6', '5', '9')): full_code = f'{plain}.SH' else: full_code = f'{plain}.SZ' st = StockInfo.get_or_none(StockInfo.code == full_code) if st and st.listing_status == 'ST': continue name = '' try: name = qmtv.getInstrumentName(plain) except Exception: pass already_in = plain in self._data.stockCodeIdMap shown += 1 cells = [ _dcell(str(r.get("score_rank", "")), _C[0][0], _C[0][1]), _dcell(f"{plain} {name}", _C[1][0], _C[1][1]), _dcell(f"{r.get("rank_predicted_rounds", 0) or 0:.4f}", _C[2][0], _C[2][1]), _dcell(f"{r.get("top_elite_prob", 0) or 0:.4f}", _C[3][0], _C[3][1]), _dcell(f"{r.get("stacking_probability", 0) or 0:.4f}", _C[4][0], _C[4][1]), ] if already_in: op_w = _C[5][0] cells.append(ft.Container( ft.Row([ft.Text('已添加', color='#AAAAAA', size=11, text_align=ft.TextAlign.CENTER)], alignment=ft.MainAxisAlignment.CENTER), width=op_w, padding=0)) else: op_w = _C[5][0] cells.append(ft.Container( ft.Row([ft.IconButton(ft.Icons.ADD, icon_size=20, tooltip='加入网格', on_click=lambda e, c=plain, n=name: self._on_add_from_scoring(c, n))], alignment=ft.MainAxisAlignment.CENTER), width=op_w, padding=0)) row = ft.Row(cells, spacing=0) rows.append(ft.Container(row, padding=ft.Padding(0, 2, 0, 2))) rows.append(ft.Divider(height=1, color='#f0f0f0')) self._scoring_table.controls = rows self._score_status.value = f"{trade_date} 候选 {shown} 只" self._score_status.color = '#4CAF50' self._score_status.update() def _on_add_from_scoring(self, stock_code: str, stock_name: str): """+ 按钮：从评分列表添加标的并打开网格配置""" tid = self._data.add_from_market(stock_code, stock_name) target = self._data.tradeTargets.get(tid) if target is None: return self._dialogs.open_config(target) self._refresh_scoring() # 刷新状态（显示"已添加"） def _on_add_from_market(self, stock_code: str): """+ 按钮：从市场监控添加标的并打开网格配置""" info = self._data.marketLog.get(stock_code) if not info: return stock_name = info.get('stock_name', '') tid = self._data.add_from_market(stock_code, stock_name) target = self._data.tradeTargets.get(tid) if target is None: return # 打开网格配置弹窗 self._dialogs.open_config(target) # ══════════════════════════════════════════════════════════════ # ActionBar — 选中行后的操作按钮栏 # ══════════════════════════════════════════════════════════════ # ══════════════════════════════════════════════════════════════ # SplashScreen # ══════════════════════════════════════════════════════════════ class _SplashScreen: def __init__(self, page: ft.Page): self._bar = ft.ProgressBar(width=340, value=0, color='#0078d4') self._status = ft.Text("正在初始化...", size=13) self._page = page page.add(ft.Container( ft.Container( ft.Column([ _text("神之一手", color='#0078d4', bold=True, size=22), _text("交易系统", color='#666666', size=14), ft.Container(height=20), self._status, ft.Container(height=8), self._bar, ], alignment=ft.MainAxisAlignment.CENTER, horizontal_alignment=ft.CrossAxisAlignment.CENTER), width=380, height=200, bgcolor=ft.Colors.SURFACE, border_radius=12, shadow=ft.BoxShadow(blur_radius=20, color='#20000000'), alignment=ft.Alignment.CENTER, ), alignment=ft.Alignment.CENTER, expand=True, bgcolor='#F5F5F5', )) page.update() def progress(self, text: str, pct: float): self._status.value = text self._bar.value = pct self._page.update() # ══════════════════════════════════════════════════════════════ # TradeDialogs # ══════════════════════════════════════════════════════════════ class _TradeDialogs: """启动/暂停/网格配置弹窗""" def __init__(self, page: ft.Page, data: _DataStore, on_done): self._page = page self._data = data self._on_done = on_done def confirm_start(self, target): dlg = ft.AlertDialog( title=_text("确认启动"), content=_text(f"确定要启动交易吗？\n\n{target.stock_code} {target.stock_name}"), actions=[ ft.TextButton("取消", on_click=lambda e: self._close()), ft.TextButton("确定", on_click=lambda e, t=target: self._do_start(t)), ], ) self._page.show_dialog(dlg) def confirm_stop(self, target): dlg = ft.AlertDialog( title=_text("确认暂停"), content=_text(f"确定要暂停交易吗？\n\n{target.stock_code} {target.stock_name}"), actions=[ ft.TextButton("取消", on_click=lambda e: self._close()), ft.TextButton("确定", on_click=lambda e, t=target: self._do_stop(t)), ], ) self._page.show_dialog(dlg) def open_config(self, target): if target.enabled: dlg = ft.AlertDialog(title=_text("提示"), content=_text("运行中的策略无法修改参数，请先暂停")) self._page.show_dialog(dlg) # 1.5 秒后自动关闭 import asyncio async def _auto_close(): await asyncio.sleep(1.5) self._page.pop_dialog() self._page.update() asyncio.ensure_future(_auto_close()) return PrintLog(LogLevel.INFO, f'[Flet-v2] 打开网格配置: {target.stock_code}') def _row(label, value, unit=''): inp = ft.TextField(value=str(value), width=120, height=30, text_size=13, content_padding=ft.Padding(6, 0, 6, 0)) return ft.Row([_text(label, size=12), inp, _text(unit, color='#888888', size=12) if unit else ft.Container(width=0)], spacing=6, alignment=ft.MainAxisAlignment.START) r1 = _row("基准价格:", target.grid_start_price, "元") r2 = _row("网格大小:", target.grid_size, "元") r3 = _row("网格交易量:", target.grid_volume, "手") r4 = _row("上方网格数量:", target.grid_upper_count, "格") r5 = _row("下方网格数量:", target.grid_lower_count, "格") r6 = _row("当前网格层级:", target.grid_index, "") all_rows = [r1, r2, r3, r4, r5, r6] inputs = [r.controls[1] for r in all_rows] # index 1 is the TextField preview = ft.Text("", size=11, italic=True) def _refresh_preview(e=None): try: bp, gs = float(inputs[0].value), float(inputs[1].value) up, lo = int(inputs[3].value), int(inputs[4].value) parts = [] for i in range(up, 0, -1): parts.append(f"{bp + gs * i:.2f}(卖{up - i + 1})") parts.append(f"→{bp:.2f}←(基准)") for i in range(1, lo + 1): p = bp - gs * i if p > 0: parts.append(f"{p:.2f}(买{i})") preview.value = " ".join(parts) or "—" except ValueError: preview.value = "" preview.update() # 实时计算：四个关键字段绑定 on_change for idx in (0, 1, 3, 4): inputs[idx].on_change = _refresh_preview def _save(e): try: target.grid_start_price = float(inputs[0].value) target.grid_size = float(inputs[1].value) target.grid_volume = int(inputs[2].value) target.grid_upper_count = int(inputs[3].value) target.grid_lower_count = int(inputs[4].value) target.grid_index = int(inputs[5].value) target.save() self._close() self._on_done() PrintLog(LogLevel.INFO, f'[Flet] 网格配置已保存: {target.targetName()}') except ValueError: pass info = ft.Column([ ft.Row([_text("股票代码:", bold=True, size=12), _text(target.stock_code, size=12)]), ft.Row([_text("股票名称:", bold=True, size=12), _text(target.stock_name, size=12)]), ], spacing=2) self._page.show_dialog(ft.AlertDialog( title=_text(f"网格配置 - {target.stock_code} ({target.stock_name})"), content=ft.Column([ ft.Container(info, bgcolor='#F5F5F5', padding=10, border_radius=6), ft.Divider(height=1, color='#e0e0e0'), *all_rows, preview, ], spacing=8, tight=True, height=420, scroll=ft.ScrollMode.AUTO), actions=[ft.TextButton("取消", on_click=lambda e: self._close()), ft.ElevatedButton("保存", on_click=_save)], )) def toast(self, msg: str): self._page.show_dialog(ft.AlertDialog( title=_text("提示"), content=_text(msg), actions=[ft.TextButton("确定", on_click=lambda e: self._close())], )) def _do_start(self, target): self._close() target.enabled = True target.save() self._data.strategyCtrl[target.get_id()] = SFGridStrategy(target) self._on_done() PrintLog(LogLevel.INFO, f'[Flet] 启动交易: {target.targetName()}') def _do_stop(self, target): self._close() target.enabled = False target.save() ctrl = self._data.strategyCtrl.pop(target.get_id(), None) if ctrl: ctrl.enabledTrading(False) self._on_done() PrintLog(LogLevel.INFO, f'[Flet] 暂停交易: {target.targetName()}') def confirm_remove(self, target): """移除网格策略 — 根据持仓量显示不同提示""" if target.current_position > 0: msg = f"该标的当前持仓 {target.current_position} 股，将移回「未分类持仓」。\n\n确定要移除网格策略吗？\n\n{target.stock_code} {target.stock_name}" else: msg = f"该标的当前无持仓，将直接删除。\n\n确定要移除吗？\n\n{target.stock_code} {target.stock_name}" dlg = ft.AlertDialog( title=_text("确认移除"), content=_text(msg), actions=[ ft.TextButton("取消", on_click=lambda e: self._close()), ft.TextButton("确定", on_click=lambda e, t=target: self._do_remove(t)), ], ) self._page.show_dialog(dlg) def _do_remove(self, target): self._close() result = self._data.remove_target(target.get_id()) self._on_done() if result == 'unclassified': PrintLog(LogLevel.INFO, f'[Flet] 降级为未分类: {target.targetName()}') elif result == 'deleted': PrintLog(LogLevel.INFO, f'[Flet] 已删除: {target.targetName()}') def _close(self): self._page.pop_dialog() self._page.update() # ══════════════════════════════════════════════════════════════ # QmtApp v2 # ══════════════════════════════════════════════════════════════ class QmtApp: """Flet 版 QMT 交易界面 v2""" def __init__(self, page: ft.Page): self.page = page self.page.title = "神之一手" self.page.theme_mode = ft.ThemeMode.LIGHT self.page.window.width = 1400 self.page.window.height = 800 self.page.padding = 0 self._data = _DataStore() self._dialogs = _TradeDialogs(page, self._data, self._refresh_ui) self._grid = _GridPanel(self._data, dialogs=self._dialogs) self._drawer = _DrawerPanel(self._data, dialogs=self._dialogs) self._prices_loaded = False self._run_startup() # ══════════════════════════════════════════════════════════ # 启动 # ══════════════════════════════════════════════════════════ def _run_startup(self): splash = _SplashScreen(self.page) asyncio.ensure_future(self._do_startup(splash)) async def _do_startup(self, splash: _SplashScreen): await asyncio.sleep(0.05) steps = [ ("正在检查 QMT 环境...", 0.10, lambda: RealQmtV._discover_qmt_port() or True), ("正在初始化交易器...", 0.35, lambda: qmtv.init_qmtv()), ("正在连接 QMT...", 0.55, lambda: qmtv.connect() or True), ("正在加载持仓数据...", 0.75, lambda: self._data.load_from_qmt()), ("正在构建界面...", 0.85, lambda: None), ("正在初始化策略...", 0.92, lambda: self._data.init_strategies()), ] for text, pct, action in steps: splash.progress(text, pct) try: if action() is False: self._show_error(f"启动失败: {text}"); return except Exception as e: self._show_error(f"启动异常: {text}\n{e}"); return splash.progress("启动完成", 1.0) await asyncio.sleep(0.3) self.page.clean() self._build_main_ui() self.page.update() self._data.pull_prices() self._data.refresh_orders() self._data.refresh_trades() self._refresh_ui() event_bus.subscribe(MarketDataUpdate, self._on_market_data) event_bus.subscribe(EventMarketActiveSwitch, self._on_market_active) event_bus.subscribe(EventTradeTargetUpdate, lambda t: self._refresh_ui()) event_bus.subscribe(EventPoolMark, self._on_pool_mark) threading.Thread(target=self._refresh_loop, daemon=True).start() # 启动股票池管理器（阶段一：T日标记） from core.sfgrid.pool_manager import PoolManager self._data._pool_manager = PoolManager() self._data._pool_manager.start() def _show_error(self, msg: str): self.page.clean() self.page.add(ft.Container( ft.Column([ ft.Icon(ft.Icons.ERROR_OUTLINE, size=48, color=ft.Colors.RED), _text(msg, size=16), ft.ElevatedButton("重试", on_click=lambda e: self._retry()), ], alignment=ft.MainAxisAlignment.CENTER, horizontal_alignment=ft.CrossAxisAlignment.CENTER), alignment=ft.Alignment.CENTER, expand=True, )) self.page.update() def _retry(self): self.page.clean() self._run_startup() # ══════════════════════════════════════════════════════════ # UI 构建 # ══════════════════════════════════════════════════════════ def _build_main_ui(self): self._sidebar_icon = _PanelIcon('sidebar', active=False, on_click=lambda e: self._toggle_drawer()) self._overlay = self._drawer.build() # 市场状态指示灯 self._market_dot = ft.Container(width=8, height=8, border_radius=4, bgcolor='#AAAAAA') self._market_label = _text("休市", color='#888888', size=12) market_status = ft.Row([self._market_dot, self._market_label], spacing=6, alignment=ft.MainAxisAlignment.CENTER) # 应用启动时同步当前状态 self._update_market_status(qmtv.isMarketActive) # 顶部标题栏 — 左侧市场状态，右侧刷新 + 侧栏按钮 title = ft.Container( ft.Row([ market_status, ft.Row([ ft.IconButton(ft.Icons.REFRESH, tooltip="刷新", icon_size=20, on_click=lambda e: self._manual_refresh()), self._sidebar_icon, ], spacing=0), ], alignment=ft.MainAxisAlignment.SPACE_BETWEEN), padding=ft.Padding(10, 10, 10, 5), ) # 网格持仓面板 — 标题 + 数据表，与右侧 drawer 风格一致 grid_body = ft.Container( ft.Column([ ft.Container(_text("网格策略持仓", bold=True, size=14), padding=ft.Padding(10, 10, 10, 5)), ft.Container(content=self._grid.build(), expand=True, padding=ft.Padding(10, 0, 10, 10)), ], expand=True), bgcolor=ft.Colors.SURFACE, expand=True, ) self.page.add(ft.Column([ title, ft.Stack([grid_body, self._overlay], expand=True), ], expand=True)) # ══════════════════════════════════════════════════════════ # UI 刷新 # ══════════════════════════════════════════════════════════ def _refresh_ui(self): self._data.refresh_orders() self._data.refresh_trades() self._grid._rebuild() self._drawer.refresh_all() self.page.update() def _manual_refresh(self): self._data.pull_prices() self._data.refresh_positions() self._data.refresh_orders() self._data.refresh_trades() self._refresh_ui() def _toggle_drawer(self): self._drawer_open = not getattr(self, '_drawer_open', False) self._overlay.visible = self._drawer_open self._sidebar_icon.set_active(self._drawer_open) self.page.update() def _hide_overlay(self): self._drawer_open = False self._overlay.visible = False self._sidebar_icon.set_active(False) self.page.update() # ══════════════════════════════════════════════════════════ # 事件 # ══════════════════════════════════════════════════════════ def _on_market_data(self, data: dict): count = self._data.apply_tick(data) if not self._prices_loaded and count > 0: self._prices_loaded = True self._refresh_ui() def _on_pool_mark(self, data: dict): """响应股票池标记事件，在日志区显示""" action = data.get('action', '') count = data.get('count', 0) codes = data.get('stock_codes', []) if action == 'scoring': msg = f'[池] 评分完成，共 {count} 只候选' elif action == 'liquidate': msg = f'[池] 沉寂检测完成，标记 {count} 只股: {", ".join(codes)}' elif action == 'eliminate': msg = f'[池] 周度淘汰检测完成，标记 {count} 只股: {", ".join(codes)}' else: msg = f'[池] 未知事件: {data}' self._data.poolLog.append(msg) # 追加到 UI 日志 if hasattr(self, '_pool_log_text') and self._pool_log_text: log = self._pool_log_text log.value = '\n'.join(self._data.poolLog[-200:]) if self._data.poolLog else '' log.update() from core.logger import PrintLog, LogLevel PrintLog(LogLevel.INFO, msg) def trigger_pool_slumber(self): """供 UI 按钮手动触发沉寂检测""" self._data._pool_manager.trigger_slumber() def trigger_pool_elim(self): """供 UI 按钮手动触淘汰检测""" self._data._pool_manager.trigger_weekly_elim() def _on_market_active(self, is_active: bool): from core.logger import PrintLog, LogLevel PrintLog(LogLevel.INFO, f'[UI] 收到市场状态切换事件 is_active={is_active}') self._update_market_status(is_active) self.page.update() PrintLog(LogLevel.INFO, f'[UI] 市场状态已更新，_market_label={self._market_label.value}') def _update_market_status(self, active: bool): """更新顶栏市场状态指示灯""" if active: self._market_dot.bgcolor = '#4CAF50' self._market_label.value = '盘中' self._market_label.color = '#2E7D32' else: self._market_dot.bgcolor = '#AAAAAA' self._market_label.value = '休市' self._market_label.color = '#888888' # ══════════════════════════════════════════════════════════ # 后台刷新 # ══════════════════════════════════════════════════════════ def _refresh_loop(self): cycle = 0 while True: time.sleep(5) cycle += 1 try: self._data.pull_prices() self._data.refresh_positions() if cycle % 6 == 0: self._data.refresh_orders() self._data.refresh_trades() self._refresh_ui() except Exception: pass # ══════════════════════════════════════════════════════════════ # PanelIcon # ══════════════════════════════════════════════════════════════ class _PanelIcon(ft.Container): _SIZE = 22; _M = 3 _C = {'bg': '#f0f0f0', 'off': '#b0b0b0', 'on': '#808080', 'active': '#0078d4'} def __init__(self, kind: str, active: bool = True, on_click=None): self._kind = kind self._active = active super().__init__(content=self._rects(), width=self._SIZE, height=self._SIZE, bgcolor=self._C['bg'], border_radius=3, ink=True, on_click=on_click, padding=ft.Padding(self._M, self._M, self._M, self._M)) def _rects(self): o, n, a = self._C['off'], self._C['on'], self._C['active'] bw, bh = 6, self._SIZE - self._M * 2 - 2 if self._kind == 'sidebar': return ft.Row([ ft.Container(width=bw, height=bh, bgcolor=n if self._active else o, border_radius=1), ft.Container(width=2), ft.Container(width=bw, height=bh, bgcolor=a if self._active else o, border_radius=1), ], spacing=0) return ft.Column([ ft.Container(width=bh, height=bw, bgcolor=n if self._active else o, border_radius=1), ft.Container(height=2), ft.Container(width=bh, height=bw, bgcolor=a if self._active else o, border_radius=1), ], spacing=0) def set_active(self, active: bool): self._active = active self.content = self._rects() # ══════════════════════════════════════════════════════════════ # 入口 # ══════════════════════════════════════════════════════════════ def main(page: ft.Page): QmtApp(page) def run(): ft.app(target=main)