信号 → 动作映射（由用户定义）：

┌──────────────────────┬────────────┬──────────────────────────────────────┐│ 图上标记              │ 输出列      │ 动作                                  │├──────────────────────┼────────────┼──────────────────────────────────────┤│ 建/买回 ↑             │ SIG_BUY     │ 买入：程序设置的数量                    ││ B（重新站上支撑）      │ SIG_REGAIN_SUPPORT │ 买入：程序设置的数量            ││ 减留底 ↓              │ SIG_REDUCE  │ 卖出：可用 - 最小保留量                 ││                       │             │   （可转债留 10 张 / 股票留 100 股）     ││ 清仓 ↓                │ SIG_CLEAR   │ 卖出：全部可用数量                      ││ S（跌破支撑）          │ SIG_BREAK_SUPPORT │ 卖出：全部可用数量                ││ 顶                    │ SIG_TOP     │ 卖出：可用的一半（向下取整到最小单位）    ││ 底                    │ SIG_BOTTOM  │ 买入：设置数量的一半（向下取整到最小单位） │└──────────────────────┴────────────┴──────────────────────────────────────┘

class BSBandStrategy:   """BS红绿带策略：指标信号 → 自动下单。"""# ── 默认参数（可被 alpha_config.json 覆盖）──  DEFAULT_KLINE_PERIOD = 'day'  DEFAULT_CONFIRM_MODE = 'live'  # live=用当前K线；close=只用已收盘的上一根  DEFAULT_KLINE_COUNT = 320  MIN_KLINE_FOR_INDICATOR = 40  # MA(V,120) + BARSLAST 需要足够历史  ERROR_LOG_INTERVAL_SECONDS = 300  # 同一标的取数失败日志节流def __init__(self, app=None):    ”””    Args:        app: QuantAppQt5 实例，用于取 K 线、持仓、执行买卖    ”””    self.app = app# 开关与参数    self._enabled = False    self.kline_period = self.DEFAULT_KLINE_PERIOD    self.confirm_mode = self.DEFAULT_CONFIRM_MODE    self.kline_count = self.DEFAULT_KLINE_COUNT    self.keep_min_override = 0# 0 = 自动（转债10 / 股票100）    self.top_bottom_half = True# 顶卖一半 / 底买一半    self.monitor_list: List[str] = []# 监控列表（留空=仅持仓）    self.cooldown_seconds = 0# 额外冷却（秒），0=只靠 K 线去重# 运行态    self._fired: Dict[str, Tuple[str, set]] = {}# code -> (bar_ts, {action,...})    self._last_action: Dict[str, dict] = {}    self._err_log_at: Dict[str, float] = {}    self._last_scan_at = 0.0# ══════════════════════════════════════════════════════════#  配置# ══════════════════════════════════════════════════════════def load_config(self, config):    ”””从 alpha_config.json 配置加载参数（键名与 qt_config_dialog 一致）。”””    if config is None:        return    get_value = config.get if hasattr(config, 'get') else (lambda k, d=None: d)    raw_enabled = get_value('是否启用BS红绿带策略', '否')    self._enabled = str(raw_enabled).strip() in ('是', 'true', 'True', '1', 'yes', 'on')    period = get_value('BS红绿带K线周期', None)    if period is not None and str(period).strip():        self.kline_period = str(period).strip()    mode = get_value('BS红绿带信号确认', None)    if mode is not None:        mode = str(mode).strip()        self.confirm_mode = 'close' if mode in ('收盘确认', '收盘', 'close') else 'live'    try:        v = get_value('BS红绿带最小K线数量', None)        if v is not None:            self.kline_count = max(self.MIN_KLINE_FOR_INDICATOR, int(float(v)))    except (TypeError, ValueError):        pass    try:        v = get_value('BS红绿带留底数量', None)        if v is not None:            self.keep_min_override = max(0, int(float(v)))    except (TypeError, ValueError):        pass    half_raw = get_value('BS红绿带顶底半仓', '是')    self.top_bottom_half = str(half_raw).strip() in ('是', 'true', 'True', '1', 'yes', 'on')    try:        v = get_value('BS红绿带信号冷却秒', None)        if v is not None:            self.cooldown_seconds = max(0, int(float(v)))    except (TypeError, ValueError):        pass# 监控列表：支持 list 或逗号串    watch = get_value('BS红绿带监控列表', None)    codes: List[str] = []    if isinstance(watch, (list, tuple)):        for item in watch:            c = str(item).strip()[:6]            if c.isdigit() and len(c) == 6:                codes.append(c)    elif watch is not None:        text = str(watch).replace('，', ',').replace(' ', ',')        for item in text.split(','):            c = item.strip()[:6]            if c.isdigit() and len(c) == 6:                codes.append(c)    self.monitor_list = sorted(set(codes))    logger.info(        '[BS红绿带] 配置加载: enabled=%s period=%s confirm=%s keep_min=%s '        'top_bottom_half=%s cooldown=%ds watch=%d只',        self._enabled, self.kline_period, self.confirm_mode,        self.keep_min_override or '自动', self.top_bottom_half,        self.cooldown_seconds, len(self.monitor_list),    )def set_enabled(self, enabled: bool):    self._enabled = bool(enabled)    logger.info('[BS红绿带] 策略已%s', '启用' if self._enabled else '停用')def is_enabled(self) -> bool:    return self._enabled# ══════════════════════════════════════════════════════════#  对外主入口# ══════════════════════════════════════════════════════════def monitor_and_trade(self, positions=None) -> list:    ”””扫描标的、判断信号并执行买卖。    Args:        positions: 持仓列表；None 时从 app.position_data 读取    Returns:        list: 本次执行的交易 [{code,name,action,price,qty,reason}]    ”””    if not self._enabled:        return []# ── 扫描节流：日线不必 10 秒一扫（K 线不会变），分钟线才需要盯 ──    now = time.time()    if self._last_scan_at and (now - self._last_scan_at) < self._min_scan_interval():        return []    self._last_scan_at = now    trades: list = []    try:        pos_map = self._position_map(positions)        codes = sorted(set(pos_map.keys()) | set(self.monitor_list))        if not codes:            return []        for code in codes:            try:                trade = self._process_code(code, pos_map.get(code) or {})                if trade:                    trades.append(trade)            except Exception as e:                self._log_throttled(code, '单标的处理异常: %s' % e, exc_info=True)    except Exception as e:        logger.error('[BS红绿带] 监控异常: %s', e, exc_info=True)    return tradesdef check_signals_only(self, positions=None) -> list:    ”””只算信号不下单（预览/调试用）。”””    signals: list = []    try:        pos_map = self._position_map(positions)        codes = sorted(set(pos_map.keys()) | set(self.monitor_list))        for code in codes:            try:                info = self._evaluate(code, pos_map.get(code) or {}, dry_run=True)                if info:                    signals.append(info)            except Exception as e:                logger.debug('[BS红绿带] 预览异常 %s: %s', code, e)    except Exception as e:        logger.warning('[BS红绿带] 预览异常: %s', e)    return signals# ══════════════════════════════════════════════════════════#  单标的处理# ══════════════════════════════════════════════════════════def _process_code(self, code: str, pos: dict) -> Optional[dict]:    decision = self._evaluate(code, pos, dry_run=False)    if not decision:        return None    action = decision['action']    qty = int(decision['qty'])    price = float(decision['price'])    name = decision['name']    reason = decision['reason']    bar_ts = decision['bar_ts']    if qty <= 0:        return None# ── K 线级去重：同一根 K 线的同一动作只做一次 ──    if self._already_fired(code, bar_ts, action):        return None# ── 单根 K 线只做一个动作：已有更高优先级动作执行过就跳过 ──    prev = self._fired.get(code)    if prev and prev[0] == bar_ts and prev[1]:        prev_best = min(prev[1], key=lambda a: ACTION_PRIORITY.get(a, 9))        if ACTION_PRIORITY.get(prev_best, 9) <= ACTION_PRIORITY.get(action, 9):            return None    ok = self._submit(code, name, action, price, qty, reason, decision)# 无论成功失败都标记，避免一根 K 线内反复重试轰炸券商    self._mark_fired(code, bar_ts, action)    if not ok:        return None    record = {        'code': code, 'name': name, 'action': action,        'action_label': ACTION_LABEL.get(action, action),        'price': price, 'qty': qty, 'reason': reason, 'bar_ts': bar_ts,        'at': time.time(),    }    self._last_action[code] = record    return recorddef _evaluate(self, code: str, pos: dict, dry_run: bool) -> Optional[dict]:    ”””计算某标的当前应执行的动作（不落单）。”””    df = self._get_kline(code)    if df is None or len(df) < self.MIN_KLINE_FOR_INDICATOR:        return None    from domain.algorithms.bs_band_main_indicator import calc_bs_band_main_indicator    indicator = calc_bs_band_main_indicator(df)    if indicator is None or len(indicator) == 0:        return None    idx = self._bar_index(len(indicator))    if idx is None:        return None    row = indicator.iloc[idx]    bar_ts = self._bar_timestamp(df, idx)    if bar_ts is None:        return None    price = float(pd.to_numeric(df['close'], errors='coerce').iloc[-1] or 0)    if not np.isfinite(price) or price <= 0:        return None    available = int(pos.get('available') or 0)    name = str(pos.get('name') or code)# 买入基础数量 = 程序设置的数量（交易模式 金额/数量 已在 App 侧解析）    base_qty = self._default_buy_qty(code, price)    action, qty, reason = self._decide(row, available, base_qty, code)    if action is None:        return None    return {        'code': code, 'name': name, 'action': action, 'qty': int(qty),        'price': price, 'reason': reason, 'bar_ts': bar_ts,        'available': available, 'base_qty': base_qty,    }def _decide(self, row, available: int, base_qty: int,            code: str) -> Tuple[Optional[str], int, str]:    ”””信号 → (动作, 数量, 原因)。按风险优先顺序判定，命中即返回。”””    lot = _lot(code)    keep = self._keep_min(code)# ① 清仓 / S（跌破支撑）：卖出全部可用    if _flag(row, 'SIG_CLEAR'):        qty = self._sellable_all(available, lot)        if qty > 0:            return ACT_SELL_ALL, qty, '清仓（MA10下穿MA20）'        return None, 0, '清仓信号但无可用数量'    if _flag(row, 'SIG_BREAK_SUPPORT'):        qty = self._sellable_all(available, lot)        if qty > 0:            return ACT_SELL_ALL, qty, 'S：跌破支撑线'        return None, 0, 'S信号但无可用数量'# ② 减留底：卖到只剩最小保留量    if _flag(row, 'SIG_REDUCE'):        qty = int((max(0, available - keep) // lot) * lot)        if qty > 0:            return ACT_SELL_REDUCE, qty, '减留底（保留%d，卖出%d）' % (keep, qty)        return None, 0, '减留底信号但可用数量不足（可用%d 保留%d）' % (available, keep)# ③ 顶：卖出可用的一半（向下取整到最小单位）    if self.top_bottom_half and _flag(row, 'SIG_TOP'):        qty = int((available // lot // 2) * lot)        if qty > 0:            return ACT_SELL_HALF, qty, '顶：卖出一半（可用%d→%d）' % (available, qty)        return None, 0, '顶信号但不足半个最小单位（可用%d 单位%d）' % (available, lot)# ④ 建仓 / 买回 / B：买入程序设置的数量    if _flag(row, 'SIG_BUY'):        if base_qty > 0:            return ACT_BUY_FULL, base_qty, '建仓/买回'        return None, 0, '建/买回信号但设置数量为0'    if _flag(row, 'SIG_REGAIN_SUPPORT'):        if base_qty > 0:            return ACT_BUY_FULL, base_qty, 'B：重新站上支撑线'        return None, 0, 'B信号但设置数量为0'# ⑤ 底：买入设置数量的一半（向下取整到最小单位）    if self.top_bottom_half and _flag(row, 'SIG_BOTTOM'):        qty = int((base_qty // lot // 2) * lot)        if qty > 0:            return ACT_BUY_HALF, qty, '底：买入一半（设置%d→%d）' % (base_qty, qty)        return None, 0, '底信号但设置数量的一半不足一个最小单位（设置%d 单位%d）' % (base_qty, lot)    return None, 0, ''# ══════════════════════════════════════════════════════════#  下单# ══════════════════════════════════════════════════════════def _submit(self, code: str, name: str, action: str, price: float,            qty: int, reason: str, decision: dict) -> bool:    ”””统一走 App.buy / App.sell。”””    if self.app is None:        return False    label = ACTION_LABEL.get(action, action)    try:        if action in (ACT_BUY_FULL, ACT_BUY_HALF):            buy_kwargs = dict(                code=code, price=price, qty=int(qty),                strategy_name='BS红绿带', signal=reason,                name=name, notify=True,            )            if action == ACT_BUY_HALF:# 「一半」必须是精确的一半：关闭金额模式重算 + 关闭情绪放大                buy_kwargs['skip_trade_mode_calib'] = True                buy_kwargs['apply_bond_sentiment'] = False            logger.warning('[BS红绿带] 买入 %s(%s) %s 价格%.3f 数量%d 原因:%s',                           name, code, label, price, qty, reason)            ok = self.app.buy(**buy_kwargs)        else:            logger.warning('[BS红绿带] 卖出 %s(%s) %s 价格%.3f 数量%d 原因:%s',                           name, code, label, price, qty, reason)            ok = self.app.sell(code=code, price=price, qty=int(qty),                               strategy_name='BS红绿带', signal=reason,                               name=name, notify=True)        return bool(ok)    except Exception as e:        logger.error('[BS红绿带] 下单异常 %s %s: %s', code, label, e, exc_info=True)        return False# ══════════════════════════════════════════════════════════#  数量/参数辅助# ══════════════════════════════════════════════════════════def _keep_min(self, code: str) -> int:    ”””最小保留量：配置优先，否则转债 10 张 / 股票 100 股。”””    if self.keep_min_override > 0:        return int(self.keep_min_override)    return _lot(code)def _min_scan_interval(self) -> float:    ”””两次扫描的最小间隔（秒）。日线 60s，分钟线 30s。”””    period = str(self.kline_period or '').strip().lower()    if period in ('day', 'daily', '1d', 'd'):        return 60.0    return 30.0@staticmethoddef _sellable_all(available: int, lot: int) -> int:    ”””全部可卖数量：按最小单位取整；不足一个最小单位时（零碎持仓）全卖。”””    available = int(available or 0)    if available <= 0:        return 0    if available < lot:        return available    return int((available // lot) * lot)def _default_buy_qty(self, code: str, price: float) -> int:    ”””程序设置的单笔买入数量（复用主程序的交易模式解析）。”””    try:        if self.app is not None and hasattr(self.app, '_default_buy_qty'):            qty = int(self.app._default_buy_qty(code, price) or 0)            return max(0, qty)    except Exception as e:        logger.debug('[BS红绿带] 读取程序买入数量失败 %s: %s', code, e)    lot = _lot(code)    return lot# ══════════════════════════════════════════════════════════#  数据获取# ══════════════════════════════════════════════════════════def _get_kline(self, code: str) -> Optional[pd.DataFrame]:    ”””取配置周期的 K 线（先 ws_client，再 data_manager 兜底）。”””    if self.app is None:        return None    df = None    try:        ws = getattr(self.app, 'ws_client', None)        if ws is not None and hasattr(ws, 'get_kline'):            df = ws.get_kline(code, period=self.kline_period, count=self.kline_count)    except Exception as e:        self._log_throttled(code, 'WS取K线失败: %s' % e)    if df is None or len(df) == 0:        try:            dm = getattr(self.app, 'data_manager', None)            if dm is not None and hasattr(dm, 'fetch_kline_data'):                df = dm.fetch_kline_data(code, period=self.kline_period,                                         count=self.kline_count)        except Exception as e:            self._log_throttled(code, 'data_manager取K线失败: %s' % e)    if df is None or len(df) == 0:        return None    return self._normalize_df(df)@staticmethoddef _normalize_df(df: pd.DataFrame) -> pd.DataFrame:    ”””列名归一化到 open/high/low/close/volume（指标函数依赖英文列名）。”””    if not isinstance(df, pd.DataFrame):        return None    rename = {}    for target, aliases in _COL_ALIASES.items():        if target in df.columns:            continue        for alias in aliases:            if alias in df.columns:                rename[alias] = target                break    out = df.rename(columns=rename) if rename else df    for col in ('open', 'high', 'low', 'close'):        if col not in out.columns:            return None    if 'volume' not in out.columns:        out = out.copy()        out['volume'] = 0.0    return out.reset_index(drop=True)def _bar_index(self, length: int) -> Optional[int]:    ”””决定用哪一根 K 线做判断。    · live  ：当前（最新）K 线 —— 盘中实时响应    · close ：倒数第二根 —— 只用已收盘的 K 线，最保守    ”””    if length <= 0:        return None    if self.confirm_mode == 'close':        return length - 2 if length >= 2 else None    return length - 1@staticmethoddef _bar_timestamp(df: pd.DataFrame, idx: int) -> Optional[str]:    ”””取 K 线唯一标识（用于去重）。同一根 K 线时间戳稳定不变。”””    for col in _TIME_COLS:        if col in df.columns:            try:                val = df[col].iloc[idx]                text = str(val).strip()                if text and text.lower() not in ('nan', 'nat', 'none'):                    return text            except Exception:                continue# 没有时间列：用行号兜底（同长度下稳定）    return 'idx:%d' % int(idx)# ══════════════════════════════════════════════════════════#  持仓解析# ══════════════════════════════════════════════════════════def _position_map(self, positions=None) -> Dict[str, dict]:    ”””持仓列表 → {code: {'name':.., 'available':..}}。”””    if positions is None and self.app is not None:        positions = getattr(self.app, 'position_data', None) or []    result: Dict[str, dict] = {}    for pos in (positions or []):        code, name, available = self._parse_position(pos)        if not code:            continue        result[code] = {'name': name, 'available': int(available)}    return result@staticmethoddef _parse_position(pos) -> Tuple[str, str, int]:    ”””解析一条持仓，返回 (code, name, available)。”””    try:        if isinstance(pos, dict):            code = str(pos.get('证券代码') or pos.get('code') or '').strip()[:6]            name = str(pos.get('证券名称') or pos.get('name') or code).strip()            available = int(float(                pos.get('可用余额') or pos.get('available') or                pos.get('可卖数量') or pos.get('可用数量') or 0))        else:            code = str(getattr(pos, 'code', '') or '').strip()[:6]            name = str(getattr(pos, 'name', '') or code).strip()            available = int(getattr(pos, 'available', 0) or 0)        if not code.isdigit() or len(code) != 6:            return '', '', 0        return code, name or code, max(0, available)    except Exception:        return '', '', 0# ══════════════════════════════════════════════════════════#  去重 / 日志# ══════════════════════════════════════════════════════════def _already_fired(self, code: str, bar_ts: str, action: str) -> bool:    ”””同一根 K 线的同一动作已执行过 → True。额外冷却期内也返回 True。”””    rec = self._fired.get(code)    if rec and rec[0] == bar_ts and action in rec[1]:        return True# 额外冷却（同一标的最近一次下单后 N 秒内不再动作）    if self.cooldown_seconds > 0:        last = self._last_action.get(code) or {}        try:            if (time.time() - float(last.get('at') or 0)) < self.cooldown_seconds:                return True        except (TypeError, ValueError):            pass    return Falsedef _mark_fired(self, code: str, bar_ts: str, action: str):    rec = self._fired.get(code)    if rec and rec[0] == bar_ts:        rec[1].add(action)        self._fired[code] = (bar_ts, rec[1])    else:        self._fired[code] = (bar_ts, {action})    if code in self._last_action:        self._last_action[code]['at'] = time.time()    else:        self._last_action[code] = {'at': time.time()}def _log_throttled(self, code: str, message: str, exc_info: bool = False):    ”””同一标的的重复错误日志按间隔节流，避免刷爆日志文件。”””    now = time.time()    last = self._err_log_at.get(code, 0.0)    if now - last < self.ERROR_LOG_INTERVAL_SECONDS:        return    self._err_log_at[code] = now    logger.warning('[BS红绿带] %s %s', code, message, exc_info=exc_info)# ══════════════════════════════════════════════════════════#  状态查询（供 UI 展示）# ══════════════════════════════════════════════════════════def get_status(self) -> dict:    return {        'enabled': self._enabled,        'period': self.kline_period,        'confirm_mode': self.confirm_mode,        'watch_count': len(self.monitor_list),        'keep_min': self.keep_min_override or 'auto',        'top_bottom_half': self.top_bottom_half,        'cooldown': self.cooldown_seconds,        'last_scan_at': self._last_scan_at,        'last_actions': dict(self._last_action),    }def reset_fired(self, code: str = None):    ”””清空去重记录（换周期/手动重置时用）。”””    if code is None:        self._fired.clear()    else:        self._fired.pop(str(code).strip()[:6], None)