""" 082 期特码分析 (纯挂牌, 不用 Qi-1 期数据) 按 user 2026-07-30 实战需求: 1. 不跑频率 (5 期数据无意义) 2. 不混 Qi-2 期号码 (081 期 已开, 不算 082 资料) 3. 只用挂牌 + 玄机诗 + 红字 推演 4. 输出 5 个候选特码 (有重点, 按挂牌共识排序) 5. 不强行 4 框架 (河洛/梅花/玄空/奇门) 公式 (reference 没推演方法) """ import re import json import sqlite3 import subprocess import datetime import os from collections import Counter 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='082'): """特码候选 - 按挂牌共识排序 规则: 1. 挂牌 彩图挂直接给号 (权重最高 × 5) 2. 挂牌 爆 生肖 (权重 × 4) 3. 玄机诗 诗象 给号 (权重 × 3) 4. 彩霸王 三一玄数 (权重 × 2) 5. 玄机字字型 (权重 × 2) """ weights = Counter() # Step 1: 综合挂牌 列表 + 详情 gua_html = curl_url('https://sol.2344a.cc/zongheguapai/') gua_links = re.findall(rf'href="(/zongheguapai/\d+\.html)"[^>]*>.*?{period}期', gua_html)[:3] for link in gua_links: f = fetch_sol_detail(link) # 正版彩图挂 给出号 if '正版彩图挂' in f: m = re.search(r'\d+', f['正版彩图挂']) if m: weights[int(m.group(0))] += 5 # 彩图挂直接给号最高权重 # 另版挂 给出号 if '另版挂' in f: m = re.search(r'\d+', f['另版挂']) if m: weights[int(m.group(0))] += 3 # 尾数 (如 1尾,3尾) -> +10, +30 if '尾数' in f: for m in re.finditer(r'(\d+)尾', f['尾数']): weights[int(m.group(1))] += 2 weights[int(m.group(1)) + 10] += 1 weights[int(m.group(1)) + 20] += 1 weights[int(m.group(1)) + 30] += 1 # 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: weights[num] += 3 # 彩霸王 三一玄数 if '三一' in line or '一三' in line: weights[3] += 2 weights[1] += 2 weights[13] += 2 weights[31] += 2 # 玄机字 (沐字型 8 划) if '《沐》' in line: weights[8] += 2 # 红马蓝狗 (马 红色 = 偏红, 狗 蓝色 = 偏蓝) if '红马' in line: weights[12] += 1 # 马=12 偏红 if '蓝狗' in line: weights[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): weights[n] += 1 return weights def format_te_ma_result(weights, period=None, budget=15, qi_week=None, qi_day=None, qi_nq=None): """输出特码候选 + 推荐分配 period: 要分析哪期 (None=自动取 v_xg.json Qi) 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, 默认用 Qi if period is None: period = qi_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-2 期已开号码 (不是 Qi 期) # 如果 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+ top = weights.most_common(8) L.append("🏆 特码候选 (按挂牌共识权重):") L.append("") L.append("| 排序 | 特码 | 权重 | 来源 |") L.append("|---|---|---|---|") # 常见号映射 (通用) src_map = { 9: '诗象 (09, 47 单出) + 彩图挂 09', 33: '彩图挂 33', 5: '彩图挂 05', 47: '诗象 (单, 单出)', 13: '彩霸王 三一玄数', 3: '彩霸王 三一', 1: '彩霸王 一三', 31: '彩霸王 一三', 8: '玄机字 (沐 8 划) + 中宫生气', } for i, (num, w) in enumerate(top, 1): src = src_map.get(num, '其他挂牌提示') 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 第一个参数) period = cmd if period is None: # 自动取 v_xg.json 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('Qi', '082') except Exception: period = '082' weights = 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, period, 15, qi_week, qi_day, qi_nq))