CodeForge
c4bb8632ca
- 新增 model_full_report_dlt.py: 大乐透专用全量报告脚本 - 统一4块结构: 回滚迭代→数据采集→推理分析→机选号码 - 统一术语: 前区/后区(大乐透) 对齐 红球/蓝球(双色球) - 统一评分: TOP10前区+TOP5后区 归一化概率输出 - 更新 SKILL.md: 双色球/大乐透统一模板+文件结构
301 lines
12 KiB
Python
301 lines
12 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
新模型集成入口：替代 lottery_tool.py 的 full_report
|
||
保持4块结构化输出，推理分析和机选号码使用6阶段数学模型
|
||
"""
|
||
|
||
import json, math, os, datetime, hashlib, random, re, sys
|
||
from collections import Counter
|
||
import numpy as np
|
||
|
||
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
|
||
DATA_DIR = os.path.join(BASE_DIR, "data")
|
||
OUTPUT_DIR = os.path.join(BASE_DIR, "model_output")
|
||
SSQ_FILE = os.path.join(DATA_DIR, "ssq_history.json")
|
||
FEEDBACK_FILE = os.path.join(DATA_DIR, "feedback.json")
|
||
CONFIG_FILE = os.path.join(DATA_DIR, "config.json")
|
||
|
||
# ================================================================
|
||
# 加载数据和模型参数
|
||
# ================================================================
|
||
data = json.load(open(SSQ_FILE, "r", encoding="utf-8"))
|
||
records = []
|
||
historical_combos = set()
|
||
for r in data:
|
||
reds = sorted([int(x) for x in r["reds"]])
|
||
blue = int(r["blue"])
|
||
if len(reds) == 6 and all(1 <= n <= 33 for n in reds) and 1 <= blue <= 16:
|
||
records.append({"issue": r["issue"], "reds": reds, "blue": blue, "date": r.get("date", "")})
|
||
historical_combos.add(tuple(reds))
|
||
|
||
records.sort(key=lambda x: int(x["issue"]))
|
||
N = len(records)
|
||
latest = records[-1]
|
||
|
||
# 加载模型参数
|
||
mp_file = os.path.join(OUTPUT_DIR, "model_parameters.json")
|
||
if os.path.exists(mp_file):
|
||
mp = json.load(open(mp_file, "r"))
|
||
MODEL_VERSION = "v1.0_6phase"
|
||
else:
|
||
mp = None
|
||
MODEL_VERSION = "v1.0_legacy"
|
||
|
||
# ================================================================
|
||
# 加载回滚统计
|
||
# ================================================================
|
||
config = json.load(open(CONFIG_FILE, "r")) if os.path.exists(CONFIG_FILE) else {"iteration_count": 0}
|
||
feedback = json.load(open(FEEDBACK_FILE, "r")) if os.path.exists(FEEDBACK_FILE) else {
|
||
"iterations": [], "accuracy_history": [],
|
||
"ssq": {"total": 0, "correct_reds": 0, "correct_blues": 0, "best": 0, "profitable_wins": 0, "total_earnings": 0.0}
|
||
}
|
||
|
||
# ================================================================
|
||
# 计算下期信息
|
||
# ================================================================
|
||
prev_issue = latest["issue"]
|
||
prev_date = latest.get("date", "")
|
||
prev_reds = latest["reds"]
|
||
prev_blue = latest["blue"]
|
||
next_issue = str(int(prev_issue) + 1)
|
||
|
||
today = datetime.datetime.now()
|
||
draw_days = [2, 4, 7] # 周二、四、日
|
||
draw_time = "21:15"
|
||
draw_hour, draw_min = map(int, draw_time.split(":"))
|
||
now_hour, now_min = today.hour, today.minute
|
||
today_draw = today.isoweekday() in draw_days
|
||
time_past = (now_hour > draw_hour) or (now_hour == draw_hour and now_min >= draw_min)
|
||
if today_draw and not time_past:
|
||
delta = 0
|
||
else:
|
||
delta = 1
|
||
while (today + datetime.timedelta(days=delta)).isoweekday() not in draw_days:
|
||
delta += 1
|
||
next_draw = today + datetime.timedelta(days=delta)
|
||
next_draw_str = next_draw.strftime("%Y-%m-%d")
|
||
wd_map = {1:"周一",2:"周二",3:"周三",4:"周四",5:"周五",6:"周六",7:"周日"}
|
||
next_wd = wd_map.get(next_draw.isoweekday(), "")
|
||
|
||
# ================================================================
|
||
# 4块输出
|
||
# ================================================================
|
||
print("\n" + "=" * 60)
|
||
print(" [1/4] 回滚迭代")
|
||
print("=" * 60)
|
||
|
||
ver = config.get("iteration_count", 0)
|
||
stats = feedback.get("ssq", {})
|
||
total_recs = stats.get("total", 0)
|
||
profitable = stats.get("profitable_wins", 0)
|
||
win_rate = "%.1f%%" % (profitable / total_recs * 100) if total_recs > 0 else "0.0%"
|
||
earnings = stats.get("total_earnings", 0.0)
|
||
|
||
print(" 当前版本号: %s (第6阶段数学建模)" % MODEL_VERSION)
|
||
print(" 版本胜率: %s (%d期推荐 / %d次盈利)" % (win_rate, total_recs, profitable))
|
||
print(" 累计中奖收益金额: RMB %.2f" % earnings)
|
||
|
||
# ==================== 第2块：数据采集 ====================
|
||
print("\n" + "=" * 60)
|
||
print(" [2/4] 数据采集")
|
||
print("=" * 60)
|
||
print(" 采集期数范围: 第%s期 ~ 第%s期（%d期）" % (records[0]["issue"], prev_issue, N))
|
||
print(" 上一期号: 第%s期" % prev_issue)
|
||
print(" 开奖日期: %s" % prev_date)
|
||
r_str = " ".join("%02d" % n for n in prev_reds)
|
||
print(" 开奖号码: 红球 %s 蓝球 %02d" % (r_str, prev_blue))
|
||
|
||
# ==================== 第3块：推理分析 ====================
|
||
print("\n" + "=" * 60)
|
||
print(" [3/4] 推理分析（%s | 数据源: 500.com）" % MODEL_VERSION)
|
||
print("=" * 60)
|
||
|
||
if mp:
|
||
# --- 6阶段模型输出 ---
|
||
p3 = mp["phase3_params"]
|
||
p4 = mp["phase4_params"]
|
||
p5 = mp["phase5_params"]
|
||
|
||
# 步骤1: 贝叶斯后验概率 TOP10
|
||
red_post = {int(k): float(v) for k, v in p3["red_posterior"].items()}
|
||
top10 = sorted(red_post.items(), key=lambda x: -x[1])[:10]
|
||
print(" >> 步骤1 - 贝叶斯后验概率: TOP10红球")
|
||
print(" %s" % " ".join("%02d(%.4f)" % (n, s) for n, s in top10))
|
||
|
||
# 步骤2: 蓝球后验 TOP5
|
||
blue_post = {int(k): float(v) for k, v in p3["blue_posterior"].items()}
|
||
top5b = sorted(blue_post.items(), key=lambda x: -x[1])[:5]
|
||
print(" >> 步骤2 - 蓝球后验概率: TOP5蓝球")
|
||
print(" %s" % " ".join("%02d(%.4f)" % (n, s) for n, s in top5b))
|
||
|
||
# 步骤3: 聚类分组
|
||
candidate = p4["candidate_pool"]
|
||
eliminated = p4["eliminated"]
|
||
print(" >> 步骤3 - K-Means聚类降维")
|
||
print(" K=%d | 轮廓系数=%.4f" % (p4["red_k"], p4["red_silhouette"]))
|
||
print(" 核心候选池(%d个): %s" % (len(candidate), " ".join("%02d" % n for n in candidate[:15]) + ("..." if len(candidate) > 15 else "")))
|
||
|
||
# 步骤4: 权重体系
|
||
sw = p5["objective_weights"]
|
||
print(" >> 步骤4 - 综合得分权重体系（6阶段数学模型）")
|
||
print(" Score = %.1f×后验 + %.1f×泊松 + %.1f×伽马遗漏" % (sw["w1_posterior"], sw["w2_poisson"], sw["w3_gamma"]))
|
||
print(" 方法链路: 泊松频次→负二项/伽马拟合→贝叶斯后验→K-Means→ILP约束求解")
|
||
else:
|
||
print(" [⚠️ 新模型参数未加载，使用基础统计]")
|
||
|
||
# ==================== 第4块：机选号码 ====================
|
||
print("\n" + "=" * 60)
|
||
print(" [4/4] 机选号码（第%s期 | %s %s）" % (next_issue, next_draw_str, next_wd))
|
||
print("=" * 60)
|
||
|
||
if mp:
|
||
# --- 6阶段模型推荐 ---
|
||
w1 = p3["score_weights"]["alpha1_posterior"]
|
||
w2 = p3["score_weights"]["alpha2_poisson"]
|
||
w3 = p3["score_weights"]["alpha3_miss"]
|
||
|
||
red_freq = {int(k): v["freq"] for k, v in mp["phase1_params"]["red_freq"].items()}
|
||
lam = mp["phase1_params"]["poisson_lambda_red"]
|
||
|
||
# 泊松权重
|
||
red_poisson = {}
|
||
for n in range(1, 34):
|
||
k = red_freq.get(n, 0)
|
||
log_p = k * math.log(lam) - lam
|
||
for i in range(1, k + 1):
|
||
log_p -= math.log(i)
|
||
red_poisson[n] = math.exp(log_p)
|
||
total_pois = sum(red_poisson.values())
|
||
red_poisson = {n: v / total_pois for n, v in red_poisson.items()}
|
||
|
||
# 遗漏权重
|
||
red_miss = mp["phase1_params"]["red_miss"]
|
||
red_gamma = {}
|
||
for n in range(1, 34):
|
||
miss = red_miss.get(str(n), {"avg_miss": 5, "last_miss": 0})
|
||
avg = miss["avg_miss"]
|
||
last = miss["last_miss"]
|
||
red_gamma[n] = min(1.0, last / (avg * 2)) if avg > 0 else 0.5
|
||
total_gam = sum(red_gamma.values())
|
||
red_gamma = {n: v / total_gam for n, v in red_gamma.items()}
|
||
|
||
# 综合得分
|
||
red_score = {}
|
||
for n in range(1, 34):
|
||
red_score[n] = w1 * red_post[n] + w2 * red_poisson[n] + w3 * red_gamma[n]
|
||
|
||
# 蓝球得分
|
||
blue_freq = {int(k): v["freq"] for k, v in mp["phase1_params"]["blue_freq"].items()}
|
||
lam_b = mp["phase1_params"]["poisson_lambda_blue"]
|
||
blue_poisson = {}
|
||
for n in range(1, 17):
|
||
k = blue_freq.get(n, 0)
|
||
log_p = k * math.log(lam_b) - lam_b
|
||
for i in range(1, k + 1):
|
||
log_p -= math.log(i)
|
||
blue_poisson[n] = math.exp(log_p)
|
||
total_pois_b = sum(blue_poisson.values())
|
||
blue_poisson = {n: v / total_pois_b for n, v in blue_poisson.items()}
|
||
|
||
blue_score = {}
|
||
for n in range(1, 17):
|
||
blue_score[n] = w1 * blue_post[n] + w2 * blue_poisson[n] + w3 * 0.5
|
||
|
||
# 约束检查
|
||
constraints = p5["constraints"]
|
||
def check_constraints(combo):
|
||
odd = sum(1 for n in combo if n % 2 == 1)
|
||
if not (constraints["odd_even"]["min_odd"] <= odd <= constraints["odd_even"]["max_odd"]):
|
||
return False
|
||
big = sum(1 for n in combo if n > 16)
|
||
if not (constraints["big_small"]["min_big"] <= big <= constraints["big_small"]["max_big"]):
|
||
return False
|
||
consec = sum(1 for i in range(len(combo) - 1) if combo[i + 1] - combo[i] == 1)
|
||
if not (constraints["consecutive"]["min_consec"] <= consec <= constraints["consecutive"]["max_consec"]):
|
||
return False
|
||
s = sum(combo)
|
||
if not (constraints["sum_range"]["min_sum"] <= s <= constraints["sum_range"]["max_sum"]):
|
||
return False
|
||
hot = sum(1 for n in combo if red_score[n] > 0.03)
|
||
if not (constraints["hot_cold"]["min_hot"] <= hot <= constraints["hot_cold"]["max_hot"]):
|
||
return False
|
||
z1 = sum(1 for n in combo if 1 <= n <= 11)
|
||
z2 = sum(1 for n in combo if 12 <= n <= 22)
|
||
z3 = sum(1 for n in combo if 23 <= n <= 33)
|
||
if not all(constraints["zone_balance"]["min_zone"] <= z <= constraints["zone_balance"]["max_zone"] for z in [z1, z2, z3]):
|
||
return False
|
||
if tuple(combo) in historical_combos:
|
||
return False
|
||
return True
|
||
|
||
# 生成推荐
|
||
import itertools
|
||
sorted_nums = sorted(range(1, 34), key=lambda n: -red_score[n])
|
||
best_combos = []
|
||
seed = hashlib.md5(("6phase_ssq_%s" % prev_issue).encode()).hexdigest()
|
||
rng = random.Random(seed)
|
||
|
||
for _ in range(30000):
|
||
weights = np.array([red_score[n] for n in range(1, 34)])
|
||
weights = weights / weights.sum()
|
||
selected = np.random.choice(range(1, 34), size=6, replace=False, p=weights)
|
||
combo = sorted(selected.tolist())
|
||
if check_constraints(combo):
|
||
score = sum(red_score[n] for n in combo)
|
||
if len(best_combos) < 10 or score > best_combos[-1][0]:
|
||
best_combos.append((score, combo))
|
||
best_combos.sort(key=lambda x: -x[0])
|
||
if len(best_combos) > 10:
|
||
best_combos = best_combos[:10]
|
||
|
||
for combo in itertools.combinations(sorted_nums[:15], 6):
|
||
combo_list = sorted(list(combo))
|
||
if check_constraints(combo_list):
|
||
score = sum(red_score[n] for n in combo_list)
|
||
if len(best_combos) < 10 or score > best_combos[-1][0]:
|
||
best_combos.append((score, combo_list))
|
||
best_combos.sort(key=lambda x: -x[0])
|
||
if len(best_combos) > 10:
|
||
best_combos = best_combos[:10]
|
||
|
||
seen = set()
|
||
unique = []
|
||
for s, c in best_combos:
|
||
if tuple(c) not in seen:
|
||
seen.add(tuple(c))
|
||
unique.append((s, c))
|
||
results = unique[:5]
|
||
|
||
blue_ranked = sorted(blue_score.items(), key=lambda x: -x[1])[:3]
|
||
primary_blue = blue_ranked[0][0]
|
||
|
||
for i, (score, combo) in enumerate(results):
|
||
r_str = " ".join("%02d" % n for n in combo)
|
||
odd = sum(1 for n in combo if n % 2 == 1)
|
||
big = sum(1 for n in combo if n > 16)
|
||
s = sum(combo)
|
||
marker = " ⭐" if i == 0 else ""
|
||
print("\n 第%d组: 红球 %s 蓝球 %02d%s" % (i + 1, r_str, primary_blue, marker))
|
||
print(" 得分:%.4f | 奇偶:%d:%d | 大小:%d:%d | 和:%d" % (
|
||
score, odd, 6 - odd, big, 6 - big, s))
|
||
|
||
# 保存推荐
|
||
analysis = json.load(open(os.path.join(DATA_DIR, "analysis_ssq.json"), "r")) if os.path.exists(os.path.join(DATA_DIR, "analysis_ssq.json")) else {}
|
||
analysis["last_recommendation"] = {
|
||
"reds": results[0][1],
|
||
"blues": [primary_blue],
|
||
"generated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
|
||
"model_version": MODEL_VERSION
|
||
}
|
||
json.dump(analysis, open(os.path.join(DATA_DIR, "analysis_ssq.json"), "w"), ensure_ascii=False, indent=2)
|
||
|
||
else:
|
||
print(" [⚠️ 新模型未加载，请先运行 model_phase1~6.py 训练模型]")
|
||
|
||
print("\n" + "=" * 60)
|
||
print(" ⚠️ 免责声明: 本推荐仅基于历史统计建模的相对概率评分，")
|
||
print(" 不预测下期开奖，不保证中奖。彩票是独立随机事件。")
|
||
print("=" * 60)
|
||
print("\n 开奖后把实际号码发我，帮你回滚迭代校正精度 🍀")
|