406 lines
15 KiB
Python
406 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
screen_whales.py — 鲸鱼筛选工具
|
|
|
|
扫描 持仓/ 目录下所有 *-blocks.csv, 累计每个地址的:
|
|
- 出场轮数 / 胜出轮数 / 胜率
|
|
- 平均仓位 (筛掉小单噪声)
|
|
- hedged 比例 (⚖️ 标记的轮次占比)
|
|
|
|
对比 tracked_whales_list.txt 给出建议:
|
|
- 新增候选 (达到阈值但不在列表)
|
|
- 剔除建议 (现有地址胜率掉到阈值下 / 出场不足 / hedged 比例过高)
|
|
|
|
用法:
|
|
python screen_whales.py # dry-run, 只打印建议
|
|
python screen_whales.py --apply # 写回 tracked_whales_list.txt
|
|
python screen_whales.py --min-rounds 30 # 自定义阈值
|
|
|
|
阈值 (CLI 可覆盖, 默认值见 parse_args):
|
|
--min-rounds 20 最少出场轮数
|
|
--min-winrate 0.85 最低胜率
|
|
--min-avg-shares 200 最低平均仓位 (排除小单噪声)
|
|
--max-hedged-ratio 0.50 hedged 轮次最大占比 (纯市商剔除)
|
|
"""
|
|
|
|
import argparse
|
|
import glob
|
|
import os
|
|
import re
|
|
import sys
|
|
from collections import defaultdict
|
|
from datetime import datetime
|
|
|
|
|
|
HOLDINGS_DIR = os.path.join(os.path.dirname(os.path.abspath(__file__)), "持仓")
|
|
WHALES_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "tracked_whales_list.txt")
|
|
|
|
|
|
def load_address_stats(holdings_dir: str) -> dict:
|
|
"""扫描 holdings_dir 下所有 *-blocks.csv, 返回 {addr: stats}
|
|
|
|
stats = {
|
|
"rounds": set(), # 出场的轮次 key (file stem)
|
|
"wins": set(), # 胜出的轮次 key
|
|
"positions": [int], # 每轮持仓数
|
|
"hedged_rounds": set(), # 标记 ⚖️ 的轮次 key
|
|
}
|
|
"""
|
|
stats = defaultdict(lambda: {
|
|
"rounds": set(),
|
|
"wins": set(),
|
|
"positions": [],
|
|
"hedged_rounds": set(),
|
|
})
|
|
|
|
files = sorted(glob.glob(os.path.join(holdings_dir, "*-blocks.csv")))
|
|
for fp in files:
|
|
fname = os.path.basename(fp)
|
|
m = re.match(r"(\d{4}-\d{4})-(up|down)-blocks\.csv", fname)
|
|
if not m:
|
|
continue
|
|
round_key, winner_side = m.group(1), m.group(2)
|
|
|
|
try:
|
|
with open(fp, "r", encoding="utf-8-sig") as f:
|
|
# DictReader 需要 headers, 但 utf-8-sig 解码后可能丢 BOM, 用 Sniffer 兜底
|
|
import csv as csv_mod
|
|
reader = csv_mod.DictReader(f)
|
|
for row in reader:
|
|
addr = (row.get("地址") or "").strip()
|
|
if not addr or not addr.startswith("0x"):
|
|
continue
|
|
addr = addr.lower() # 统一小写, 与 tracked_whales_list.txt 对齐
|
|
side = (row.get("方向") or "").strip().lower()
|
|
try:
|
|
shares = float(row.get("持仓") or 0)
|
|
except ValueError:
|
|
shares = 0
|
|
hedged_field = row.get("对冲") or ""
|
|
is_hedged = "⚖" in hedged_field
|
|
|
|
s = stats[addr]
|
|
s["rounds"].add(round_key)
|
|
s["positions"].append(shares)
|
|
if side == winner_side:
|
|
s["wins"].add(round_key)
|
|
if is_hedged:
|
|
s["hedged_rounds"].add(round_key)
|
|
except Exception as e:
|
|
print(f"⚠️ 读取 {fname} 失败: {e}", file=sys.stderr)
|
|
|
|
return stats
|
|
|
|
|
|
def load_existing_whales(path: str) -> dict:
|
|
"""读 tracked_whales_list.txt, 返回 {addr: tier_label}
|
|
|
|
tier_label = 该地址所在 section 的标题行 (如 "TIER S: 50+ rounds...")
|
|
"""
|
|
result = {}
|
|
current_tier = "(未分组)"
|
|
if not os.path.exists(path):
|
|
return result
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
for line in f:
|
|
line_stripped = line.strip()
|
|
if not line_stripped:
|
|
current_tier = "(未分组)"
|
|
continue
|
|
if line_stripped.startswith("#"):
|
|
# section 标题: "# TIER S: 50+ rounds, 90%+ WR, large capital"
|
|
m = re.match(r"#\s*(TIER\s+\w+:.*)", line_stripped)
|
|
if m:
|
|
current_tier = m.group(1)
|
|
continue
|
|
if line_stripped.startswith("0x"):
|
|
addr = line_stripped[:42].lower()
|
|
result[addr] = current_tier
|
|
return result
|
|
|
|
|
|
def evaluate(stats: dict, existing: dict, args) -> list:
|
|
"""根据阈值评估每个地址, 返回 [(addr, metrics)] 列表 (按胜率 desc)
|
|
|
|
已有鲸鱼 (tracked_whales_list.txt 里) 与新候选用不同门槛:
|
|
- 已有: 跳过轮数阈值 (外部已验证), 仅查胜率/均仓/hedged
|
|
- 新增: 走完整 min_rounds + 胜率 + 均仓 + hedged
|
|
|
|
metrics = {
|
|
"rounds", "wins", "winrate", "avg_shares",
|
|
"hedged_ratio", "passes": bool, "fail_reasons": [str],
|
|
"in_existing": bool,
|
|
}
|
|
"""
|
|
results = []
|
|
for addr, s in stats.items():
|
|
rounds_n = len(s["rounds"])
|
|
wins_n = len(s["wins"])
|
|
winrate = wins_n / rounds_n if rounds_n else 0
|
|
avg_shares = sum(s["positions"]) / len(s["positions"]) if s["positions"] else 0
|
|
hedged_ratio = len(s["hedged_rounds"]) / rounds_n if rounds_n else 0
|
|
is_existing = addr in existing
|
|
|
|
fail_reasons = []
|
|
# 已有鲸鱼: 轮数阈值仅作"数据不足"的提示, 不算 fail
|
|
if rounds_n < args.min_rounds:
|
|
if is_existing:
|
|
# 已有鲸鱼: 标记但不剔除 (外部已用更大样本验证)
|
|
fail_reasons.append(f"本机轮数{rounds_n}<{args.min_rounds} (跳过, 外部已验证)")
|
|
else:
|
|
fail_reasons.append(f"轮数{rounds_n}<{args.min_rounds}")
|
|
|
|
# 胜率门槛对所有都生效 (但允许最小轮数放宽)
|
|
# 已有: 本机轮数 < 5 时不判 (数据太少, 跳过胜率检查)
|
|
if winrate < args.min_winrate:
|
|
if is_existing and rounds_n >= 5:
|
|
fail_reasons.append(f"胜率{winrate:.0%}<{args.min_winrate:.0%}")
|
|
elif not is_existing:
|
|
fail_reasons.append(f"胜率{winrate:.0%}<{args.min_winrate:.0%}")
|
|
|
|
# 均仓: 对所有生效, 但已有鲸鱼放宽 (外部可能已验证过)
|
|
if avg_shares < args.min_avg_shares:
|
|
if is_existing:
|
|
fail_reasons.append(f"均仓{avg_shares:.0f}<{args.min_avg_shares} (跳过, 外部已验证)")
|
|
else:
|
|
fail_reasons.append(f"均仓{avg_shares:.0f}<{args.min_avg_shares}")
|
|
|
|
# hedged 比例: 严格, 所有都生效
|
|
if hedged_ratio > args.max_hedged_ratio:
|
|
fail_reasons.append(f"hedged{hedged_ratio:.0%}>{args.max_hedged_ratio:.0%}")
|
|
|
|
# "跳过" 标记的 reason 不算真正的 fail (只是提示)
|
|
real_fails = [r for r in fail_reasons if "(跳过" not in r]
|
|
results.append({
|
|
"addr": addr,
|
|
"rounds": rounds_n,
|
|
"wins": wins_n,
|
|
"winrate": winrate,
|
|
"avg_shares": avg_shares,
|
|
"hedged_ratio": hedged_ratio,
|
|
"passes": not real_fails,
|
|
"fail_reasons": fail_reasons,
|
|
"real_fails": real_fails,
|
|
"in_existing": is_existing,
|
|
})
|
|
|
|
results.sort(key=lambda r: (r["winrate"], r["rounds"]), reverse=True)
|
|
return results
|
|
|
|
|
|
def diff(existing: dict, results: list, args) -> dict:
|
|
"""对比已有列表 + 评估结果, 返回建议动作
|
|
|
|
已有鲸鱼 (in_existing=True): 只在有真 fail (real_fails) 时才建议剔除
|
|
新候选: passes 才加入
|
|
|
|
returns {
|
|
"add": [addr], # 达到阈值但不在列表
|
|
"remove": [(addr, reasons)],
|
|
"keep": [addr],
|
|
"watch": [(addr, metrics)], # 接近阈值但失败
|
|
}
|
|
"""
|
|
add = []
|
|
remove = []
|
|
keep = []
|
|
watch = []
|
|
|
|
for r in results:
|
|
addr = r["addr"]
|
|
if r["in_existing"]:
|
|
# 已有: 有真 fail 才剔除
|
|
if r["real_fails"]:
|
|
remove.append((addr, r["real_fails"]))
|
|
else:
|
|
keep.append(addr)
|
|
else:
|
|
# 新增: passes 才入选
|
|
if r["passes"]:
|
|
add.append(addr)
|
|
else:
|
|
# 观察名单: 轮数够但其他指标接近
|
|
if r["rounds"] >= args.min_rounds * 0.7:
|
|
watch.append(r)
|
|
|
|
return {"add": add, "remove": remove, "keep": keep, "watch": watch}
|
|
|
|
|
|
def print_report(diff_result: dict, results: list, existing: dict, args):
|
|
"""打印人类可读的报告"""
|
|
print(f"\n{'='*70}")
|
|
print(f"鲸鱼筛选报告 ({datetime.now().strftime('%Y-%m-%d %H:%M')})")
|
|
print(f"{'='*70}")
|
|
print(f"扫描持仓/: {sum(1 for _ in results)} 个地址")
|
|
print(f"tracked_whales_list.txt: {len(existing)} 个地址")
|
|
print(f"\n阈值: rounds≥{args.min_rounds} | winrate≥{args.min_winrate:.0%} | "
|
|
f"均仓≥{args.min_avg_shares} | hedged≤{args.max_hedged_ratio:.0%}")
|
|
|
|
# ── 剔除建议 ──
|
|
print(f"\n📛 剔除建议 ({len(diff_result['remove'])} 个):")
|
|
if diff_result["remove"]:
|
|
for addr, reasons in diff_result["remove"][:20]:
|
|
print(f" {addr[:10]}…{addr[-4:]} {'; '.join(reasons)}")
|
|
if len(diff_result["remove"]) > 20:
|
|
print(f" …还有 {len(diff_result['remove']) - 20} 个")
|
|
else:
|
|
print(" (无)")
|
|
|
|
# ── 新增候选 ──
|
|
print(f"\n✅ 新增候选 ({len(diff_result['add'])} 个):")
|
|
if diff_result["add"]:
|
|
# 按胜率+轮数排序
|
|
add_metrics = {r["addr"]: r for r in results}
|
|
sorted_add = sorted(diff_result["add"],
|
|
key=lambda a: (add_metrics[a]["winrate"],
|
|
add_metrics[a]["rounds"]),
|
|
reverse=True)
|
|
for addr in sorted_add[:20]:
|
|
m = add_metrics[addr]
|
|
print(f" {addr[:10]}…{addr[-4:]} "
|
|
f"胜率{m['winrate']:.0%}({m['wins']}/{m['rounds']}) "
|
|
f"均仓{m['avg_shares']:.0f} hedged={m['hedged_ratio']:.0%}")
|
|
if len(sorted_add) > 20:
|
|
print(f" …还有 {len(sorted_add) - 20} 个")
|
|
else:
|
|
print(" (无)")
|
|
|
|
# ── 观察名单 (接近阈值) ──
|
|
near_threshold = [r for r in diff_result["watch"]
|
|
if r["rounds"] >= args.min_rounds * 0.7
|
|
and r["winrate"] >= args.min_winrate - 0.1][:10]
|
|
if near_threshold:
|
|
print(f"\n👀 接近阈值 ({len(near_threshold)} 个, 仅参考):")
|
|
for r in near_threshold:
|
|
print(f" {r['addr'][:10]}…{r['addr'][-4:]} "
|
|
f"胜率{r['winrate']:.0%}({r['wins']}/{r['rounds']}) "
|
|
f"均仓{r['avg_shares']:.0f} hedged={r['hedged_ratio']:.0%} "
|
|
f"→ {'; '.join(r['fail_reasons'])}")
|
|
|
|
|
|
def apply_changes(diff_result: dict, results: list, existing: dict,
|
|
whales_file: str, args):
|
|
"""写回 tracked_whales_list.txt
|
|
|
|
策略: 保留原文件所有注释 + tier section 标题, 只更新地址行
|
|
新增地址放 "AUTO-SCREENED" section, 剔除的从原 section 删除
|
|
"""
|
|
import csv as csv_mod
|
|
|
|
with open(whales_file, "r", encoding="utf-8") as f:
|
|
original_lines = f.readlines()
|
|
|
|
remove_addrs = {addr for addr, _ in diff_result["remove"]}
|
|
add_addrs = set(diff_result["add"])
|
|
add_metrics = {r["addr"]: r for r in results}
|
|
|
|
new_lines = []
|
|
in_auto_section = False
|
|
for line in original_lines:
|
|
stripped = line.strip()
|
|
# 跳过被剔除的地址行
|
|
if stripped.startswith("0x") and stripped[:42].lower() in remove_addrs:
|
|
continue
|
|
# 检测 "AUTO" section 边界
|
|
if "AUTO" in stripped and stripped.startswith("#"):
|
|
in_auto_section = True
|
|
new_lines.append(line)
|
|
|
|
# 追加 AUTO section (如果之前没有)
|
|
today = datetime.now().strftime("%Y-%m-%d")
|
|
auto_section = [
|
|
f"\n# ============================================\n",
|
|
f"# AUTO-SCREENED: {today} (run screen_whales.py --apply)\n",
|
|
f"# 阈值: rounds≥{args.min_rounds} winrate≥{args.min_winrate:.0%} "
|
|
f"均仓≥{args.min_avg_shares} hedged≤{args.max_hedged_ratio:.0%}\n",
|
|
]
|
|
# 检查文件末尾是否已有 AUTO section
|
|
has_auto = any("AUTO-SCREENED" in l for l in new_lines)
|
|
|
|
if add_addrs:
|
|
if not has_auto:
|
|
new_lines.extend(auto_section)
|
|
else:
|
|
new_lines.append(f"\n# --- {today} 更新 ---\n")
|
|
sorted_add = sorted(add_addrs,
|
|
key=lambda a: (add_metrics[a]["winrate"],
|
|
add_metrics[a]["rounds"]),
|
|
reverse=True)
|
|
for addr in sorted_add:
|
|
m = add_metrics[addr]
|
|
new_lines.append(
|
|
f"{addr} # {m['winrate']:.0%} ({m['wins']}/{m['rounds']}) "
|
|
f"均仓{m['avg_shares']:.0f}\n"
|
|
)
|
|
|
|
# 写回
|
|
with open(whales_file, "w", encoding="utf-8") as f:
|
|
f.writelines(new_lines)
|
|
|
|
print(f"\n✅ 已更新 {whales_file}")
|
|
print(f" 剔除: {len(remove_addrs)} 个 | 新增: {len(add_addrs)} 个")
|
|
print(f" ⚠️ onchain_leaderboard.py 将在下个回合自动热重载")
|
|
|
|
|
|
def parse_args():
|
|
p = argparse.ArgumentParser(
|
|
description="扫描持仓CSV, 重算鲸鱼胜率, 建议新增/剔除",
|
|
formatter_class=argparse.RawTextHelpFormatter,
|
|
)
|
|
p.add_argument("--apply", action="store_true",
|
|
help="写回 tracked_whales_list.txt (默认 dry-run)")
|
|
p.add_argument("--yes", "-y", action="store_true",
|
|
help="跳过确认提示 (用于自动化/CI)")
|
|
p.add_argument("--min-rounds", type=int, default=20,
|
|
help="最少出场轮数 (默认 20)")
|
|
p.add_argument("--min-winrate", type=float, default=0.85,
|
|
help="最低胜率 0~1 (默认 0.85)")
|
|
p.add_argument("--min-avg-shares", type=float, default=200,
|
|
help="最低平均持仓 (排除小单噪声, 默认 200)")
|
|
p.add_argument("--max-hedged-ratio", type=float, default=0.50,
|
|
help="hedged 轮次最大占比 (默认 0.50, 纯市商剔除)")
|
|
return p.parse_args()
|
|
|
|
|
|
def main():
|
|
# Windows GBK console 兼容
|
|
try:
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
except Exception:
|
|
pass
|
|
|
|
args = parse_args()
|
|
|
|
if not os.path.isdir(HOLDINGS_DIR):
|
|
print(f"❌ 持仓目录不存在: {HOLDINGS_DIR}", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
print(f"📂 扫描 {HOLDINGS_DIR} ...")
|
|
stats = load_address_stats(HOLDINGS_DIR)
|
|
print(f" 发现 {len(stats)} 个独立地址")
|
|
|
|
existing = load_existing_whales(WHALES_FILE)
|
|
results = evaluate(stats, existing, args)
|
|
d = diff(existing, results, args)
|
|
print_report(d, results, existing, args)
|
|
|
|
if args.apply:
|
|
if d["remove"] or d["add"]:
|
|
print(f"\n⚠️ 即将修改 {WHALES_FILE}:")
|
|
print(f" 剔除 {len(d['remove'])} 个 | 新增 {len(d['add'])} 个")
|
|
if args.yes:
|
|
apply_changes(d, results, existing, WHALES_FILE, args)
|
|
else:
|
|
resp = input(" 确认? [y/N]: ").strip().lower()
|
|
if resp == "y":
|
|
apply_changes(d, results, existing, WHALES_FILE, args)
|
|
else:
|
|
print(" 取消")
|
|
else:
|
|
print("\n(无变更, 跳过写入)")
|
|
else:
|
|
print(f"\n(dry-run, 加 --apply 才会写入文件)")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main() |