mike
657dc41c46
- OKX交易自动化 (okx-auto-position, okx-crypto, okx-exchange) - 交易信号处理 (signal-confirmation-templates, trading-signal-aggregator) - 量化因子挖掘 (quant-factor-mining) - 长桥集成 (longbridge-cli, longbridge-python-sdk) - 六合彩分析 (lottery-hk) - 股息投资 (dividend-investing, dividend-scanner) - 日内交易 (intraday-trading) - 同花顺 (tonghuashun)
406 lines
14 KiB
Python
406 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
香港六合彩开奖抓取与分析（SQLite版）
|
|
数据源: https://tktk.tktk4.cc/ww.htm (天空彩票)
|
|
|
|
用法:
|
|
python3 lottery.py fetch # 抓取最新开奖结果
|
|
python3 lottery.py add <期号> <号码> # 手动添加
|
|
python3 lottery.py add_full <期号> <号码> <生肖> # 带生肖添加
|
|
python3 lottery.py history [期数] # 查看历史记录
|
|
python3 lottery.py analyze # 分析（频率/热号/冷号/生肖）
|
|
python3 lottery.py zodiac # 生肖属性表
|
|
python3 lottery.py next # 下期开奖时间
|
|
python3 lottery.py import_json <文件> # 导入JSON历史数据
|
|
"""
|
|
|
|
import json, os, sys, re, sqlite3
|
|
from datetime import datetime
|
|
from collections import Counter
|
|
|
|
DATA_DIR = os.path.expanduser("~/.hermes/trading")
|
|
DB_FILE = os.path.join(DATA_DIR, "lottery.db")
|
|
|
|
# 生肖对照表（网站实际映射，从浏览器071期数据验证）
|
|
ZODIAC_BY_MOD = {
|
|
0: "狗", 1: "猪", 2: "蛇", 3: "马", 4: "羊", 5: "虎",
|
|
6: "兔", 7: "鼠", 8: "牛", 9: "猴", 10: "鸡", 11: "龙"
|
|
}
|
|
|
|
# 五行对照表
|
|
ELEMENT_MAP = {
|
|
1: "木", 2: "木", 3: "火", 4: "火", 5: "土", 6: "土",
|
|
7: "金", 8: "金", 9: "水", 10: "水", 11: "木", 12: "木",
|
|
13: "火", 14: "火", 15: "土", 16: "土", 17: "金", 18: "金",
|
|
19: "水", 20: "水", 21: "木", 22: "木", 23: "火", 24: "火",
|
|
25: "土", 26: "土", 27: "金", 28: "金", 29: "水", 30: "水",
|
|
31: "木", 32: "木", 33: "火", 34: "火", 35: "土", 36: "土",
|
|
37: "金", 38: "金", 39: "水", 40: "水", 41: "木", 42: "木",
|
|
43: "火", 44: "火", 45: "土", 46: "土", 47: "金", 48: "金",
|
|
49: "水",
|
|
}
|
|
|
|
# 波色对照表
|
|
COLOR_MAP = {
|
|
"红波": [1, 2, 7, 8, 12, 13, 18, 19, 23, 24, 29, 30, 34, 35, 40, 45, 46],
|
|
"蓝波": [3, 4, 9, 10, 14, 15, 20, 25, 26, 31, 36, 37, 41, 42, 47, 48],
|
|
"绿波": [5, 6, 11, 16, 17, 21, 22, 27, 28, 32, 33, 38, 39, 43, 44, 49],
|
|
}
|
|
|
|
def get_zodiac(num):
|
|
return ZODIAC_BY_MOD.get(num % 12, "?")
|
|
|
|
def get_element(num):
|
|
return ELEMENT_MAP.get(num, "?")
|
|
|
|
def get_color(num):
|
|
for color, nums in COLOR_MAP.items():
|
|
if num in nums:
|
|
return color
|
|
return "未知"
|
|
|
|
def get_db():
|
|
os.makedirs(DATA_DIR, exist_ok=True)
|
|
conn = sqlite3.connect(DB_FILE)
|
|
conn.row_factory = sqlite3.Row
|
|
conn.execute("""CREATE TABLE IF NOT EXISTS draws (
|
|
period TEXT PRIMARY KEY,
|
|
date TEXT,
|
|
n1 INTEGER, n2 INTEGER, n3 INTEGER, n4 INTEGER, n5 INTEGER, n6 INTEGER,
|
|
special INTEGER,
|
|
z1 TEXT, z2 TEXT, z3 TEXT, z4 TEXT, z5 TEXT, z6 TEXT, z_special TEXT,
|
|
e1 TEXT, e2 TEXT, e3 TEXT, e4 TEXT, e5 TEXT, e6 TEXT, e_special TEXT,
|
|
c1 TEXT, c2 TEXT, c3 TEXT, c4 TEXT, c5 TEXT, c6 TEXT, c_special TEXT,
|
|
created_at TEXT DEFAULT CURRENT_TIMESTAMP
|
|
)""")
|
|
conn.execute("""CREATE TABLE IF NOT EXISTS cold_data (
|
|
key TEXT PRIMARY KEY,
|
|
content TEXT,
|
|
updated_at TEXT DEFAULT CURRENT_TIMESTAMP
|
|
)""")
|
|
conn.commit()
|
|
return conn
|
|
|
|
def add_draw(period, numbers, zodiacs=None):
|
|
"""添加开奖结果"""
|
|
conn = get_db()
|
|
|
|
existing = conn.execute("SELECT period FROM draws WHERE period=?", (period,)).fetchone()
|
|
if existing:
|
|
print(f"⚠️ 第{period}期已存在，跳过")
|
|
conn.close()
|
|
return
|
|
|
|
nums = [int(n) for n in numbers[:6]]
|
|
special = int(numbers[6]) if len(numbers) > 6 else 0
|
|
|
|
zodiac_list = []
|
|
element_list = []
|
|
color_list = []
|
|
all_nums = nums + [special]
|
|
|
|
for i, num in enumerate(all_nums):
|
|
if zodiacs and i < len(zodiacs):
|
|
zodiac_list.append(zodiacs[i])
|
|
else:
|
|
zodiac_list.append(get_zodiac(num))
|
|
element_list.append(get_element(num))
|
|
color_list.append(get_color(num))
|
|
|
|
conn.execute("""INSERT INTO draws
|
|
(period, date, n1,n2,n3,n4,n5,n6,special, z1,z2,z3,z4,z5,z6,z_special, e1,e2,e3,e4,e5,e6,e_special, c1,c2,c3,c4,c5,c6,c_special)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(period, datetime.now().strftime("%Y-%m-%d"),
|
|
*nums, special,
|
|
*zodiac_list,
|
|
*element_list,
|
|
*color_list))
|
|
conn.commit()
|
|
conn.close()
|
|
|
|
nums_str = " ".join([f"{n:02d}" for n in nums])
|
|
print(f"✅ 第{period}期已添加: {nums_str} + {special:02d}")
|
|
|
|
def show_history(limit=10):
|
|
"""显示历史记录"""
|
|
conn = get_db()
|
|
rows = conn.execute("SELECT * FROM draws ORDER BY CAST(period AS INTEGER) DESC LIMIT ?", (limit,)).fetchall()
|
|
conn.close()
|
|
|
|
if not rows:
|
|
print("📭 暂无历史记录")
|
|
return
|
|
|
|
print(f"📋 最近{len(rows)}期开奖记录：\n")
|
|
for r in rows:
|
|
nums = [r['n1'], r['n2'], r['n3'], r['n4'], r['n5'], r['n6']]
|
|
special = r['special']
|
|
zodiacs = [r['z1'], r['z2'], r['z3'], r['z4'], r['z5'], r['z6']]
|
|
elements = [r['e1'], r['e2'], r['e3'], r['e4'], r['e5'], r['e6']]
|
|
|
|
nums_str = " ".join([f"{n:02d}" for n in nums])
|
|
print(f"第{r['period']}期 ({r['date']}): {nums_str} + {special:02d}")
|
|
|
|
info = []
|
|
for i in range(6):
|
|
info.append(f"{nums[i]:02d}({zodiacs[i]}/{elements[i]})")
|
|
info.append(f"+ {special:02d}({r['z_special']}/{r['e_special']})特")
|
|
print(f" {' '.join(info)}")
|
|
print()
|
|
|
|
def analyze():
|
|
"""分析开奖数据"""
|
|
conn = get_db()
|
|
rows = conn.execute("SELECT * FROM draws").fetchall()
|
|
conn.close()
|
|
|
|
if not rows:
|
|
print("📭 暂无数据")
|
|
return
|
|
|
|
print(f"📊 共{len(rows)}期数据分析：\n")
|
|
|
|
all_nums = []
|
|
special_nums = []
|
|
zodiac_counter = Counter()
|
|
element_counter = Counter()
|
|
color_counter = Counter()
|
|
|
|
for r in rows:
|
|
nums = [r['n1'], r['n2'], r['n3'], r['n4'], r['n5'], r['n6']]
|
|
all_nums.extend(nums)
|
|
special_nums.append(r['special'])
|
|
|
|
for i in range(1, 7):
|
|
zodiac_counter[r[f'z{i}']] += 1
|
|
element_counter[r[f'e{i}']] += 1
|
|
color_counter[r[f'c{i}']] += 1
|
|
zodiac_counter[r['z_special']] += 1
|
|
element_counter[r['e_special']] += 1
|
|
color_counter[r['c_special']] += 1
|
|
|
|
freq = Counter(all_nums)
|
|
special_freq = Counter(special_nums)
|
|
|
|
print("🔥 热号（出现最多）:")
|
|
for num, count in freq.most_common(10):
|
|
print(f" {num:02d} ({get_zodiac(num)}): {count}次")
|
|
|
|
print("\n❄️ 冷号（出现最少）:")
|
|
for num, count in freq.most_common()[-10:]:
|
|
print(f" {num:02d} ({get_zodiac(num)}): {count}次")
|
|
|
|
print("\n🎯 特码频率:")
|
|
for num, count in special_freq.most_common(10):
|
|
print(f" {num:02d} ({get_zodiac(num)}): {count}次")
|
|
|
|
print("\n🐉 生肖频率:")
|
|
for zodiac, count in zodiac_counter.most_common():
|
|
print(f" {zodiac}: {count}次")
|
|
|
|
print("\n🌊 五行频率:")
|
|
for element, count in element_counter.most_common():
|
|
print(f" {element}: {count}次")
|
|
|
|
print("\n🎨 波色频率:")
|
|
for color, count in color_counter.most_common():
|
|
print(f" {color}: {count}次")
|
|
|
|
big = sum(1 for n in all_nums if n >= 25)
|
|
small = sum(1 for n in all_nums if n < 25)
|
|
odd = sum(1 for n in all_nums if n % 2 == 1)
|
|
even = sum(1 for n in all_nums if n % 2 == 0)
|
|
print(f"\n📏 大小: 大{big} / 小{small}")
|
|
print(f"📏 单双: 单{odd} / 双{even}")
|
|
|
|
def save_cold_data(key, content):
|
|
"""保存冷数据到数据库"""
|
|
conn = get_db()
|
|
conn.execute("""INSERT OR REPLACE INTO cold_data (key, content, updated_at)
|
|
VALUES (?, ?, ?)""", (key, content, datetime.now().isoformat()))
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"✅ 冷数据已保存: {key}")
|
|
|
|
def get_cold_data(key):
|
|
"""获取冷数据"""
|
|
conn = get_db()
|
|
row = conn.execute("SELECT content FROM cold_data WHERE key=?", (key,)).fetchone()
|
|
conn.close()
|
|
return row['content'] if row else None
|
|
|
|
def import_json(filepath):
|
|
"""导入JSON历史数据到SQLite"""
|
|
with open(filepath) as f:
|
|
data = json.load(f)
|
|
|
|
conn = get_db()
|
|
count = 0
|
|
for item in data:
|
|
period = item.get('period')
|
|
if not period:
|
|
continue
|
|
|
|
existing = conn.execute("SELECT period FROM draws WHERE period=?", (period,)).fetchone()
|
|
if existing:
|
|
continue
|
|
|
|
numbers = item.get('numbers', [])
|
|
special = item.get('special', 0)
|
|
zodiacs_raw = [d.get('zodiac') for d in item.get('details', [])]
|
|
|
|
if len(numbers) < 6:
|
|
continue
|
|
|
|
all_nums = numbers[:6] + [special]
|
|
zodiac_list = []
|
|
element_list = []
|
|
color_list = []
|
|
|
|
for i, num in enumerate(all_nums):
|
|
if zodiacs_raw and i < len(zodiacs_raw) and zodiacs_raw[i]:
|
|
zodiac_list.append(zodiacs_raw[i])
|
|
else:
|
|
zodiac_list.append(get_zodiac(num))
|
|
element_list.append(get_element(num))
|
|
color_list.append(get_color(num))
|
|
|
|
conn.execute("""INSERT INTO draws
|
|
(period, date, n1,n2,n3,n4,n5,n6,special, z1,z2,z3,z4,z5,z6,z_special, e1,e2,e3,e4,e5,e6,e_special, c1,c2,c3,c4,c5,c6,c_special)
|
|
VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)""",
|
|
(period, item.get('date', ''),
|
|
*numbers[:6], special,
|
|
*zodiac_list, *element_list, *color_list))
|
|
count += 1
|
|
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"✅ 导入{count}条记录")
|
|
|
|
def zodiac_table():
|
|
"""显示生肖属性表"""
|
|
print("🐉 2026年生肖号码对照表（网站实际映射）：\n")
|
|
zodiac_nums = {}
|
|
for num in range(1, 50):
|
|
zodiac = get_zodiac(num)
|
|
if zodiac not in zodiac_nums:
|
|
zodiac_nums[zodiac] = []
|
|
zodiac_nums[zodiac].append(num)
|
|
for zodiac in ["鼠", "牛", "虎", "兔", "龙", "蛇", "马", "羊", "猴", "鸡", "狗", "猪"]:
|
|
nums = zodiac_nums.get(zodiac, [])
|
|
print(f" {zodiac}: {', '.join([f'{n:02d}' for n in nums])}")
|
|
|
|
def next_draw():
|
|
conn = get_db()
|
|
r = conn.execute("SELECT * FROM draws ORDER BY CAST(period AS INTEGER) DESC LIMIT 1").fetchone()
|
|
conn.close()
|
|
|
|
if r:
|
|
nums = [r['n1'], r['n2'], r['n3'], r['n4'], r['n5'], r['n6']]
|
|
print(f"📊 最新开奖: 第{r['period']}期")
|
|
print(f" 号码: {' '.join([f'{n:02d}' for n in nums])} + {r['special']:02d}")
|
|
print(f"\n⏰ 下期开奖时间: 每周二、四、六 21:30")
|
|
print(f" 数据源: https://tktk.tktk4.cc/ww.htm")
|
|
|
|
def save_image_link(category, title, url, period=None):
|
|
"""保存图片链接到数据库"""
|
|
conn = get_db()
|
|
conn.execute("INSERT INTO image_links (category, title, url, period) VALUES (?,?,?,?)",
|
|
(category, title, url, period))
|
|
conn.commit()
|
|
conn.close()
|
|
print(f"✅ 图片链接已保存: {category}/{title}")
|
|
|
|
def list_image_links(category=None, period=None):
|
|
"""列出图片链接"""
|
|
conn = get_db()
|
|
sql = "SELECT * FROM image_links WHERE 1=1"
|
|
params = []
|
|
if category:
|
|
sql += " AND category=?"
|
|
params.append(category)
|
|
if period:
|
|
sql += " AND period=?"
|
|
params.append(period)
|
|
sql += " ORDER BY created_at DESC LIMIT 50"
|
|
rows = conn.execute(sql, params).fetchall()
|
|
conn.close()
|
|
if not rows:
|
|
print("📭 暂无图片链接")
|
|
return
|
|
for r in rows:
|
|
print(f"[{r['category']}] {r['title']}: {r['url']}")
|
|
|
|
if __name__ == "__main__":
|
|
if len(sys.argv) < 2:
|
|
print(__doc__)
|
|
sys.exit(0)
|
|
|
|
cmd = sys.argv[1]
|
|
|
|
if cmd == "fetch":
|
|
print("⚠️ 页面使用Vue.js动态加载，开奖时间段外(21:14-21:40)可能无数据")
|
|
print(" 建议开奖期间用浏览器抓取，或用 add 命令手动添加")
|
|
|
|
elif cmd == "history":
|
|
limit = int(sys.argv[2]) if len(sys.argv) > 2 else 10
|
|
show_history(limit)
|
|
|
|
elif cmd == "analyze":
|
|
analyze()
|
|
|
|
elif cmd == "zodiac":
|
|
zodiac_table()
|
|
|
|
elif cmd == "next":
|
|
next_draw()
|
|
|
|
elif cmd == "add":
|
|
if len(sys.argv) < 4:
|
|
print("用法: python3 lottery.py add <期号> <号码,逗号分隔>")
|
|
sys.exit(1)
|
|
add_draw(sys.argv[2], [int(x) for x in sys.argv[3].split(",")])
|
|
|
|
elif cmd == "add_full":
|
|
if len(sys.argv) < 5:
|
|
print("用法: python3 lottery.py add_full <期号> <号码> <生肖>")
|
|
sys.exit(1)
|
|
add_draw(sys.argv[2], [int(x) for x in sys.argv[3].split(",")], sys.argv[4].split(","))
|
|
|
|
elif cmd == "import_json":
|
|
if len(sys.argv) < 3:
|
|
print("用法: python3 lottery.py import_json <文件路径>")
|
|
sys.exit(1)
|
|
import_json(sys.argv[2])
|
|
|
|
elif cmd == "save_cold":
|
|
if len(sys.argv) < 4:
|
|
print("用法: python3 lottery.py save_cold <key> <content>")
|
|
sys.exit(1)
|
|
save_cold_data(sys.argv[2], sys.argv[3])
|
|
|
|
elif cmd == "get_cold":
|
|
if len(sys.argv) < 3:
|
|
print("用法: python3 lottery.py get_cold <key>")
|
|
sys.exit(1)
|
|
content = get_cold_data(sys.argv[2])
|
|
if content:
|
|
print(content)
|
|
else:
|
|
print("📭 无数据")
|
|
|
|
elif cmd == "save_image":
|
|
if len(sys.argv) < 5:
|
|
print("用法: python3 lottery.py save_image <类别> <标题> <URL> [期号]")
|
|
sys.exit(1)
|
|
period = sys.argv[5] if len(sys.argv) > 5 else None
|
|
save_image_link(sys.argv[2], sys.argv[3], sys.argv[4], period)
|
|
|
|
elif cmd == "list_images":
|
|
category = sys.argv[2] if len(sys.argv) > 2 else None
|
|
list_image_links(category)
|
|
|
|
else:
|
|
print(f"未知命令: {cmd}")
|
|
print(__doc__)
|