enginestart/internal-testing-node

1import streamlit as st2from collections import deque3 4# --- 核心逻辑类 ---5class PureTradingSystem:6    def __init__(self, name, offset, n_value=4):7        self.name = name8        self.offset = offset9        self.n_value = n_value10        self.layer = 111        self.max_layer = 112        self.wins_pool = {i: 0 for i in range(1, 101)}13        self.patterns = ["OOO", "OOX", "OXX", "OXO", "XXX", "XXO", "XOO", "XOX"]14        self.queue = deque(self.patterns)15        self.last_idx = -116 17    def process_logic(self, history):18        # 截取对应路数的有效数据19        data = history[self.offset:]20        # 每3个字符为一组进行采样21        groups = [data[i:i+3] for i in range(0, len(data)-2, 3)]22        23        # 增量处理逻辑24        if len(groups) <= self.last_idx:25            self.reset_state()26            27        for i in range(self.last_idx + 1, len(groups)):28            new_g = groups[i]29            blind = self.queue[0]  # 当前盲门30            31            if new_g == blind:32                # 命中盲门：升层33                self.layer += 134                self.max_layer = max(self.max_layer, self.layer)35            else:36                # 未命中：计算赚取的单位 (1, 2, 或 3)37                earned = 038                for pos, (r, b) in enumerate(zip(new_g, blind)):39                    if r != b:40                        earned = pos + 141                        break42                self.wins_pool[self.layer] += earned43                44                # 检查是否满足降层条件45                while self.layer > 1 and self.wins_pool[self.layer] >= self.n_value:46                    overflow = self.wins_pool[self.layer] - self.n_value47                    self.wins_pool[self.layer] = 048                    self.layer -= 149                    self.wins_pool[self.layer] += overflow50            51            # 更新盲门队列 (先进先出)52            if new_g in self.queue:53                self.queue.remove(new_g)54            self.queue.append(new_g)55            self.last_idx = i56 57    def reset_state(self):58        self.__init__(self.name, self.offset, self.n_value)59 60# --- 界面配置 ---61st.set_page_config(page_title="逻辑验收终端", layout="wide")62st.title("🛡️ 逻辑执行终端 (Beta 验收版)")63st.caption("状态：算法同步运行中 | 仅供内部功能验收使用 | 请勿外传")64 65# 初始化 Session State (防止页面刷新数据丢失)66if "history" not in st.session_state:67    st.session_state.history = ""68    st.session_state.lines = [69        PureTradingSystem("A路采样", 0), 70        PureTradingSystem("B路采样", 1), 71        PureTradingSystem("C路采样", 2)72    ]73 74# --- 输入控制区 ---75c1, c2, c3, c4 = st.columns(4)76with c1: 77    if st.button("录入 庄 (O)", use_container_width=True):78        st.session_state.history += "O"79with c2: 80    if st.button("录入 闲 (X)", use_container_width=True):81        st.session_state.history += "X"82with c3: 83    if st.button("撤回上一步 (R)", use_container_width=True):84        st.session_state.history = st.session_state.history[:-1]85with c4: 86    if st.button("数据清空重置 (S)", use_container_width=True): 87        st.session_state.history = ""88        for l in st.session_state.lines:89            l.reset_state()90 91st.divider()92 93# --- 逻辑计算与结果展示 ---94total_o_weight, total_x_weight = 0, 095cols = st.columns(3)96 97for i, l in enumerate(st.session_state.lines):98    l.process_logic(st.session_state.history)99    100    # 计算当前路数的下注位置101    # 路数偏移量决定了当前是这组3个位置中的哪一个102    pos_in_group = (len(st.session_state.history) - l.offset) % 3103    # 盲门取反逻辑104    suggested_side = "X" if l.queue[0][pos_in_group] == "O" else "O"105    current_bet_val = l.layer * (pos_in_group + 1)106    107    if suggested_side == "O":108        total_o_weight += current_bet_val109    else:110        total_x_weight += current_bet_val111    112    with cols[i]:113        st.subheader(f"{l.name}")114        st.metric("实时层级", f"L{l.layer}", f"历史最高 L{l.max_layer}")115        st.write(f"当前锁定盲门: **{l.queue[0]}**")116        st.write(f"本路预测建议: **{suggested_side}** ({current_bet_val}U)")117 118# --- 综合决策区 ---119diff = total_o_weight - total_x_weight120if diff > 0:121    final_side = "庄 (O)"122    color = "red"123elif diff < 0:124    final_side = "闲 (X)"125    color = "blue"126else:127    final_side = "观望"128    color = "gray"129 130st.divider()131st.header(f"📢 综合策略建议：:{color}[{final_side}] :{color}[{abs(diff)} 单位]")132st.info(f"当前流水序列: {st.session_state.history}")