""" 088 期特码分析 (纯挂牌, 纯挂牌, 不混 Qi 期数据 (Qi 期已开, 用挂牌做前瞻)) 按 user 2026-07-30 实战需求: 1. 不跑频率 (5 期数据无意义) 2. 不混 Qi 期号码 (Qi 期已开, 用挂牌做前瞻, 不算 Nq 资料) 3. 只用挂牌 + 玄机诗 + 红字 推演 4. 输出 5 个候选特码 (有重点, 按挂牌共识排序) 5. 不强行 4 框架 (河洛/梅花/玄空/奇门) 公式 (reference 没推演方法) 🚨 默认期号必须用 v_xg.json Nq (未开下期), 不是 Qi (已开) (2026-08-13 修过, 详见 SKILL.md 硬规则 4 + 9) 改 main() default period 时只动 Nq + fallback, 别碰 Qi 字段注释 """ import re import json import sqlite3 import subprocess import datetime import os from collections import Counter, defaultdict DB_PATH = os.path.expanduser('~/.hermes/trading/lottery.db') def save_analysis(period, candidates, budget, v_xg_state): """把分析结果存 SQLite, 用于历史回顾 + 中奖检查 candidates: list of dict [{'num':16,'weight':3,'amount':5},...] v_xg_state: dict {'Qi':083,'Nq':084,'Week':'周二','Day':'04'} """ try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''INSERT INTO analysis (period, candidates, budget, v_xg_qi, v_xg_nq, v_xg_week, v_xg_day) VALUES (?, ?, ?, ?, ?, ?, ?)''', (period, json.dumps(candidates, ensure_ascii=False), budget, v_xg_state.get('Qi'), v_xg_state.get('Nq'), v_xg_state.get('Week'), v_xg_state.get('Day'))) conn.commit() conn.close() return True except Exception as e: print(f"[warn] save_analysis 失败: {e}", file=__import__('sys').stderr) return False def check_hits(): """对所有未中奖的 analysis 行, 跟 draws 表真开彩对比, 更新 hit + actual_special 真开彩存在 = draws.special, analysis.period = draws.period 用 actual_special IS NULL 找未更新行 (hit=0 是已检查, hit IS NULL 是没检查) """ try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() # 未更新 (actual_special IS NULL) 的 analysis c.execute('''SELECT a.id, a.period, a.candidates FROM analysis a WHERE a.actual_special IS NULL''') rows = c.fetchall() for aid, period, candidates_json in rows: # 查真开彩 c.execute('SELECT special FROM draws WHERE period = ?', (period,)) d = c.fetchone() if d is None or d[0] is None or d[0] == 0: continue # 还没开 或 special 未填 actual = d[0] # 查 candidates 里有没有 actual candidates = json.loads(candidates_json) hit = 1 if any(c.get('num') == actual for c in candidates) else 0 c.execute('UPDATE analysis SET actual_special=?, hit=? WHERE id=?', (actual, hit, aid)) conn.commit() conn.close() except Exception as e: print(f"[warn] check_hits 失败: {e}", file=__import__('sys').stderr) def add_draw(period, special, n1=0, n2=0, n3=0, n4=0, n5=0, n6=0): """手填真开彩: lottery_特码.py add 083 34 1 2 3 4 5 6 先 INSERT/UPDATE draws, 再 reset 该 period 所有 analysis 的 actual_special=NULL 这样 check_hits 会重新跑 (hit=0 也会被覆盖) """ try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''INSERT OR REPLACE INTO draws (period, n1, n2, n3, n4, n5, n6, special) VALUES (?, ?, ?, ?, ?, ?, ?, ?)''', (period, n1, n2, n3, n4, n5, n6, special)) # reset 该 period 所有 analysis 的 actual_special=NULL c.execute('UPDATE analysis SET actual_special=NULL, hit=NULL WHERE period = ?', (period,)) conn.commit() conn.close() # 立即 check_hits check_hits() return True except Exception as e: print(f"[ERR] add_draw 失败: {e}", file=__import__('sys').stderr) return False def list_analysis(limit=10): """列出最近 N 条分析记录 + hit 状态""" try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''SELECT id, period, candidates, budget, v_xg_qi, v_xg_nq, actual_special, hit, created_at FROM analysis ORDER BY id DESC LIMIT ?''', (limit,)) print(f"{'ID':<4} {'期号':<6} {'Top候选':<35} {'¥':<3} {'v_xg':<10} {'真开':<6} {'中':<3} {'时间':<20}") print("-" * 110) for row in c.fetchall(): aid, period, cands, budget, vq, vn, actual, hit, ts = row c_list = json.loads(cands) top = '/'.join(f"{c['num']}({c['weight']})" for c in c_list[:5]) v_xg_s = f"Qi{vq}/Nq{vn}" if vq else '?' hit_s = '✓' if hit == 1 else ('✗' if hit == 0 else '-') actual_s = str(actual) if actual is not None else '-' print(f"{aid:<4} {period:<6} {top[:35]:<35} {budget:<3} {v_xg_s:<10} {actual_s:<6} {hit_s:<3} {ts}") conn.close() except Exception as e: print(f"[ERR] list_analysis 失败: {e}", file=__import__('sys').stderr) def curl_url(url): try: result = subprocess.run( ['curl', '-x', 'http://127.0.0.1:7890', '-L', '-s', url], capture_output=True, text=True, timeout=20 ) return result.stdout except Exception as e: return f"[error: {e}]" def fetch_sol_list(path, marker, limit=10): html = curl_url(f'https://sol.2344a.cc{path}') out = [] for line in html.split('\n'): if marker in line: clean = re.sub(r'<[^>]+>', ' ', line).strip() m = re.search(rf'{marker}[^"]*?(?=]*>.*?', '', html, flags=re.DOTALL) text = re.sub(r']*>.*?', '', text, flags=re.DOTALL) text = re.sub(r'<[^>]+>', ' ', text) text = re.sub(r'\s+', ' ', text).strip() fields = {} for kw in ['正版彩图挂', '另版挂', '四字', '六肖', '尾数', '火烧', '爆', '出肖', '挂牌出肖', '挂牌成语']: m = re.search(kw + r'[:：]([^。\s]{1,30})', text) if m: fields[kw] = m.group(1).strip() return fields def get_te_ma_candidates(period='088'): """特码候选 - 按挂牌共识排序 规则: 1. 挂牌 彩图挂直接给号 (权重最高 × 5) 2. 挂牌 爆 生肖 (权重 × 4) 3. 玄机诗 诗象 给号 (权重 × 3) 4. 彩霸王 三一玄数 (权重 × 2) 5. 玄机字字型 (权重 × 2) """ weights = Counter() # 同步记录每个号码的来源 (用于输出表格) src_tracker = defaultdict(list) def add(n, w, src): if 1 <= n <= 49: weights[n] += w if src not in src_tracker[n]: src_tracker[n].append(src) # Step 1: 综合挂牌 列表 + 详情 # 2026-08-20 改: 多抓几页 (3 → 8), 挂牌稀疏时也能拿到更多数据源 gua_html = curl_url('https://sol.2344a.cc/zongheguapai/') gua_links = re.findall(rf'href="(/zongheguapai/\d+\.html)"[^>]*>.*?{period}期', gua_html)[:8] for link in gua_links: f = fetch_sol_detail(link) # 正版彩图挂 给出号 if '正版彩图挂' in f: m = re.search(r'\d+', f['正版彩图挂']) if m: add(int(m.group(0)), 5, '挂牌 彩图挂直接给号') # 另版挂 给出号 if '另版挂' in f: m = re.search(r'\d+', f['另版挂']) if m: add(int(m.group(0)), 3, f'挂牌 另版挂 ({f["另版挂"]})') # 尾数 (如 1尾,3尾) -> +10, +30 if '尾数' in f: tail_str = f['尾数'] for m in re.finditer(r'(\d+)尾', tail_str): base = int(m.group(1)) add(base, 2, f'挂牌 尾数 {base}尾 (本)') add(base + 10, 1, f'挂牌 尾数 {base}尾 (+10)') add(base + 20, 1, f'挂牌 尾数 {base}尾 (+20)') add(base + 30, 1, f'挂牌 尾数 {base}尾 (+30)') # 火烧 (特码生肖提示 — 数字解码: 加 1 个本年生肖号 +2) # 2026-08-20 改: 解码本年生肖, 让挂牌稀疏时多一个号 (单条+2 不至于污染权重) SX_TO_BENSHENG = { # 本年生肖 → 该生肖的"本数" (单数尾数对应) '鼠': 23, '牛': 22, '虎': 21, '兔': 20, '龙': 19, '蛇': 18, '马': 17, '羊': 16, '猴': 15, '鸡': 14, '狗': 13, '猪': 12, } if '火烧' in f: for sx, num in SX_TO_BENSHENG.items(): if sx in f['火烧']: add(num, 2, f'挂牌 火烧 {sx}={num} (本年生肖)') break # 出肖 / 六肖 (2026-08-20 改: 解码所有提到的生肖本数, 各 +1) for kw in ['出肖', '六肖']: if kw in f: sxs = re.findall(r'[鼠牛虎兔龙蛇羊猴鸡狗猪]', f[kw]) for sx in sxs: num = SX_TO_BENSHENG.get(sx) if num: add(num, 1, f'挂牌 {kw} {sx}={num}') # Step 2: 玄机诗 (xuanjiziliao) xuan = fetch_sol_list('/xuanjiziliao/', f'{period}期', limit=20) for line in xuan: # 诗象: 09、47 if '提供' in line or '猜' in line: for m in re.finditer(r'[（(](\d+)[)）]', line): num = int(m.group(1)) if 1 <= num <= 49: add(num, 3, f'玄机诗 诗象 ({line[:30]}...)') # 彩霸王 三一玄数 if '三一' in line or '一三' in line: for n in [3, 1, 13, 31]: add(n, 2, '玄机诗 彩霸王 三一玄数') # 玄机字 (沐字型 8 划) if '《沐》' in line: add(8, 2, '玄机字 (沐 8 划)') # 红马蓝狗 (马 红色 = 偏红, 狗 蓝色 = 偏蓝) if '红马' in line: add(12, 1, '玄机诗 红马 (马=12 偏红)') if '蓝狗' in line: add(18, 1, '玄机诗 蓝狗 (狗=18 偏蓝)') # Step 3: 六信红字 (红字 大数偏多) hong = fetch_sol_list('/lxhz/', f'{period}期', limit=5) for line in hong: if '頤養' in line or '天年' in line: # 大数偏多 (>25) for n in range(25, 50): add(n, 1, '六信红字 大数偏多 (頤養/天年)') return weights, src_tracker def format_te_ma_result(weights, src_tracker=None, period=None, budget=15, qi_week=None, qi_day=None, qi_nq=None): """输出特码候选 + 推荐分配 period: 要分析哪期 (None=自动取 v_xg.json Nq 未开) qi_week / qi_day / qi_nq: v_xg.json 字段 (None=自动取) """ # 一次性查 v_xg.json 拿所有字段 try: import urllib.request req = urllib.request.Request('https://btc.tktk.app/data/v_xg.json', headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=10) as r: v = json.loads(r.read().decode('utf-8')) qi_xg = v.get('Qi', '???') nq_xg = v.get('Nq', '???') week_xg = v.get('Week', '?') day_xg = v.get('Day', '?') except Exception: qi_xg = nq_xg = '???' week_xg = day_xg = '?' # 如果用户没传 period, 默认取 Nq (未开下期, 跟 main() 一致) # (2026-08-13 修: 之前注释写"默认用 Qi"是错的, Qi=已开期跟分析目的相反) if period is None: period = nq_xg if qi_week is None: qi_week = week_xg if qi_day is None: qi_day = day_xg if qi_nq is None: qi_nq = nq_xg L = [] L.append("=" * 62) L.append(f"{period} 期特码分析 (纯挂牌, 不混 Qi 期已开数据)") # v_xg.json 字段语义 (2026-08-04 修正): # Qi = 最新已开 (刚开) # Nq = 未开下期 # Week/Day = Nq 期开彩日 # Data.1-7 = Qi 期真开彩号码 (2026-08-04 验证) # 如果 period == Nq 期, Week/Day 就是 period 开彩日 (直接用) # 如果 period == Qi 期, Week/Day 是 Nq 期开彩日 (要 -3 天推算 Qi 期) if str(period) == str(qi_nq): # period 是 Nq 期 (未开), Week/Day 直接是 period 开彩日 L.append(f"v_xg.json: Qi={qi_xg} (最新已开) | Nq={nq_xg} (未开下期) | Week={week_xg} (Day={day_xg}) = 本期 {period} 期开彩日") elif str(period) == str(qi_xg): # period 是 Qi 期 (已开), Week/Day 是 Nq 期, 要 -3 天推算 Qi 期开彩日 L.append(f"v_xg.json: Qi={qi_xg} (最新已开) | Nq={nq_xg} (未开下期) | Week={week_xg} (Day={day_xg}) = Nq 期开彩日") L.append(f"本期 ({period}) 开彩日: Nq 期开彩日 ({qi_week} Day={qi_day}) 之前 1 个开彩日 (-3 天)") else: # period 是其他期 (Cli 传用户指定) L.append(f"v_xg.json: Qi={qi_xg} (最新已开) | Nq={nq_xg} (未开下期) | Week={week_xg} (Day={day_xg}) = Nq 期开彩日") L.append(f"本期 ({period}) 开彩日: 用户指定期 (非 Qi/Nq), 不推算") L.append("=" * 62) L.append("") L.append("📋 挂牌资料源:") L.append(" 1. https://sol.2344a.cc/zongheguapai/ (综合挂牌)") L.append(" 2. https://sol.2344a.cc/xuanjiziliao/ (玄机诗)") L.append(" 3. https://sol.2344a.cc/lxhz/ (六信红字)") L.append("") L.append(f"💰 总预算: {budget} 元 (按重点分配)") L.append("") # 排序 Top 5+ # 2026-08-20 改: 保证至少 4-5 个候选。如果 top<4, 兜底补足 (标记来源为占位) top = weights.most_common(8) if len(top) < 4: # most_common 不返回权重 0 的号, 这里显式补足到 5 个 (权重 0 = 占位,排最后) need = 5 - len(top) existing = {n for n, _ in top} for n in range(1, 50): if n not in existing: top.append((n, 0)) existing.add(n) if len(top) >= 5: break L.append("🏆 特码候选 (按挂牌共识权重):") L.append("") L.append("| 排序 | 特码 | 权重 | 来源 |") L.append("|---|---|---|---|") # 来源映射 — 优先用 src_tracker (实际抓取), 再 fallback 硬编码 def get_src(num, w): if w == 0: return '兜底占位 (挂牌稀疏)' if src_tracker and num in src_tracker and src_tracker[num]: # 显示前 2 个来源, 多了写 ... srcs = src_tracker[num][:2] txt = ' + '.join(srcs) if len(src_tracker[num]) > 2: txt += f' (+{len(src_tracker[num]) - 2})' return txt return '其他挂牌提示' for i, (num, w) in enumerate(top, 1): src = get_src(num, w) emoji = ['🥇', '🥈', '🥉', '4', '5', '6', '7', '8'][i-1] L.append(f"| {emoji} | **{num}** | {w} | {src} |") L.append("") # 重点分配 (按权重比例) L.append(f"💸 重点分配 ({budget} 元):") L.append("") L.append("| 特码 | 金额 | 占比 | 权重比 | 中奖得 |") L.append("|---|---|---|---|---|") # 按权重比例分配 total_w = sum(w for _, w in top[:5]) splits = [5, 4, 3, 2, 1] # 默认 5/4/3/2/1 分配 for i, (num, w) in enumerate(top[:5], 1): amt = splits[i-1] if i <= len(splits) else 1 win = amt * 42 emoji = ['🥇', '🥈', '🥉', '4', '5'][i-1] L.append(f"| {emoji} {num} | ¥{amt} | {amt*100//budget}% | {w}/{total_w} | ¥{win} |") L.append("") L.append(f"📊 总投入: ¥{sum(splits[:min(5, len(top))])}") L.append("") L.append("⚠️ 文化娱乐参考, 不要按这些号买") L.append("📌 按 SKILL.md 提示: 推算仅供娱乐, 不构成投注建议") return '\n'.join(L) if __name__ == '__main__': import sys cmd = sys.argv[1] if len(sys.argv) > 1 else None # CLI: add [n1 n2 n3 n4 n5 n6] if cmd == 'add': period = sys.argv[2] special = int(sys.argv[3]) nums = [int(x) for x in sys.argv[4:10]] # 最多 6 个平码 while len(nums) < 6: nums.append(0) if add_draw(period, special, *nums): print(f"✅ {period} 期真开彩: special={special} 平码={nums[:6]}") # 显示对应该期的所有 analysis 是否中 try: conn = sqlite3.connect(DB_PATH) c = conn.cursor() c.execute('''SELECT id, candidates, hit FROM analysis WHERE period = ? ORDER BY id DESC''', (period,)) for aid, cands, hit in c.fetchall(): cs = json.loads(cands) hit_num = next((c2['num'] for c2 in cs if c2['num'] == special), None) print(f" analysis #{aid}: hit={'✓' if hit==1 else '✗'} (猜 {special}: {'在Top' if hit_num else '不在Top'})") conn.close() except Exception: pass sys.exit(0) # CLI: list [N] if cmd == 'list': limit = int(sys.argv[2]) if len(sys.argv) > 2 else 10 list_analysis(limit) sys.exit(0) # CLI: hits (单独跑 check_hits) if cmd == 'hits': check_hits() print("✅ check_hits 跑完") sys.exit(0) # 默认: 跑分析 (period 第一个参数) # 🚨 默认取 v_xg.json Nq (未开下期), 不是 Qi (已开) — 见 SKILL.md 硬规则 4 period = cmd if period is None: # 自动取 v_xg.json Nq (未开下期, 不是 Qi 已开) try: import urllib.request req = urllib.request.Request('https://btc.tktk.app/data/v_xg.json', headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=10) as r: period = json.loads(r.read().decode('utf-8')).get('Nq', '088') except Exception: period = '088' weights, src_tracker = get_te_ma_candidates(period) # 自动取 v_xg.json Qi + Week + Day + Nq 一次, 传所有 # (period 保留用户 CLI 传的, 不要被 Qi 覆盖 — Qi 只用来推算开彩日) try: import urllib.request req = urllib.request.Request('https://btc.tktk.app/data/v_xg.json', headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req, timeout=10) as r: v = json.loads(r.read().decode('utf-8')) qi_week = v.get('Week', '?') qi_day = v.get('Day', '?') qi_nq = v.get('Nq', '?') qi_xg = v.get('Qi', '?') v_xg_state = {'Qi': qi_xg, 'Nq': qi_nq, 'Week': qi_week, 'Day': qi_day} except Exception: qi_week = qi_day = qi_nq = qi_xg = '?' v_xg_state = {} # Top 5 + 重点金额 (存 SQLite 用) top5 = weights.most_common(5) splits = [5, 4, 3, 2, 1] candidates_for_db = [] for i, (num, w) in enumerate(top5, 1): amt = splits[i-1] if i <= len(splits) else 1 candidates_for_db.append({'num': num, 'weight': w, 'amount': amt}) # 存 SQLite save_analysis(period, candidates_for_db, 15, v_xg_state) # 对所有未中奖的 analysis 更新 hit (开彩后) check_hits() print(format_te_ma_result(weights, src_tracker, period, 15, qi_week, qi_day, qi_nq))