132 lines
4.3 KiB
Python
132 lines
4.3 KiB
Python
"""큐레이터용 후보 가공 — 여러 엔진 결과를 하나로 병합, 중복 제거, 피처 계산."""
|
|
from typing import Dict, List, Any
|
|
from . import db
|
|
from .recommender import recommend_numbers, recommend_with_heatmap
|
|
from .analyzer import get_statistical_report
|
|
|
|
|
|
LOW_HIGH_CUT = 22 # 1~22 저구간, 23~45 고구간
|
|
|
|
|
|
def compute_features(numbers: List[int], hot: set, cold: set) -> Dict[str, Any]:
|
|
nums = sorted(numbers)
|
|
odd = sum(1 for n in nums if n % 2 == 1)
|
|
low = sum(1 for n in nums if n <= LOW_HIGH_CUT)
|
|
buckets = [0, 0, 0, 0, 0] # 1-10, 11-20, 21-30, 31-40, 41-45
|
|
for n in nums:
|
|
if n <= 10: buckets[0] += 1
|
|
elif n <= 20: buckets[1] += 1
|
|
elif n <= 30: buckets[2] += 1
|
|
elif n <= 40: buckets[3] += 1
|
|
else: buckets[4] += 1
|
|
consecutive = any(nums[i+1] - nums[i] == 1 for i in range(len(nums) - 1))
|
|
return {
|
|
"odd_count": odd,
|
|
"even_count": 6 - odd,
|
|
"low_count": low,
|
|
"high_count": 6 - low,
|
|
"range_distribution": buckets,
|
|
"has_consecutive": consecutive,
|
|
"hot_number_count": len(set(nums) & hot),
|
|
"cold_number_count": len(set(nums) & cold),
|
|
"sum": sum(nums),
|
|
}
|
|
|
|
|
|
def _key(numbers: List[int]) -> str:
|
|
return ",".join(str(n) for n in sorted(numbers))
|
|
|
|
|
|
def collect_candidates(n: int, hot: set, cold: set) -> List[Dict[str, Any]]:
|
|
"""여러 엔진에서 후보를 모으고 중복을 제거. 최대 n세트 반환.
|
|
|
|
우선순위: simulation best_picks → meta → heatmap → statistics
|
|
"""
|
|
seen = {}
|
|
sources_order = []
|
|
|
|
# 1. simulation best_picks
|
|
for row in db.get_best_picks(limit=n):
|
|
numbers = row.get("numbers") or []
|
|
if not numbers:
|
|
continue
|
|
k = _key(numbers)
|
|
if k not in seen:
|
|
seen[k] = {"numbers": sorted(numbers), "source": "simulation"}
|
|
sources_order.append(k)
|
|
|
|
# 2. meta-strategy (smart)
|
|
try:
|
|
from .generator import generate_smart_recommendation
|
|
meta = generate_smart_recommendation(sets=n)
|
|
for s in meta.get("sets", []):
|
|
numbers = s.get("numbers") or []
|
|
k = _key(numbers)
|
|
if k not in seen and numbers:
|
|
seen[k] = {"numbers": sorted(numbers), "source": "meta"}
|
|
sources_order.append(k)
|
|
except Exception:
|
|
pass
|
|
|
|
# 3. heatmap
|
|
try:
|
|
hm = recommend_with_heatmap(count=n)
|
|
for numbers in hm:
|
|
k = _key(numbers)
|
|
if k not in seen and numbers:
|
|
seen[k] = {"numbers": sorted(numbers), "source": "heatmap"}
|
|
sources_order.append(k)
|
|
except Exception:
|
|
pass
|
|
|
|
# 4. statistics
|
|
try:
|
|
st = recommend_numbers(count=n)
|
|
for numbers in st:
|
|
k = _key(numbers)
|
|
if k not in seen and numbers:
|
|
seen[k] = {"numbers": sorted(numbers), "source": "statistics"}
|
|
sources_order.append(k)
|
|
except Exception:
|
|
pass
|
|
|
|
out = []
|
|
for k in sources_order[:n]:
|
|
item = seen[k]
|
|
item["features"] = compute_features(item["numbers"], hot, cold)
|
|
out.append(item)
|
|
return out
|
|
|
|
|
|
def build_context(hot_limit: int = 3, cold_limit: int = 3) -> Dict[str, Any]:
|
|
"""주간 맥락 패키지."""
|
|
report = get_statistical_report()
|
|
latest = db.get_latest_draw()
|
|
freq = report.get("frequency", {})
|
|
sorted_freq = sorted(freq.items(), key=lambda x: -x[1])
|
|
hot = [int(k) for k, _ in sorted_freq[:hot_limit]]
|
|
|
|
sorted_cold = sorted(freq.items(), key=lambda x: x[1])
|
|
cold = [int(k) for k, _ in sorted_cold[:cold_limit]]
|
|
|
|
last_summary = ""
|
|
if latest:
|
|
nums = [latest.get(f"drwtNo{i}") for i in range(1, 7)]
|
|
odd = sum(1 for n in nums if n and n % 2 == 1)
|
|
low = sum(1 for n in nums if n and n <= LOW_HIGH_CUT)
|
|
last_summary = f"{latest['drwNo']}회: {', '.join(str(n) for n in nums)} (홀{odd}짝{6-odd}, 저{low}고{6-low})"
|
|
|
|
my_perf = []
|
|
try:
|
|
from .purchase_manager import get_recent_performance
|
|
my_perf = get_recent_performance(limit=3)
|
|
except Exception:
|
|
my_perf = []
|
|
|
|
return {
|
|
"hot_numbers": hot,
|
|
"cold_numbers": cold,
|
|
"last_draw_summary": last_summary,
|
|
"my_recent_performance": my_perf,
|
|
}
|