k
87 lines
3.1 KiB
Python
87 lines
3.1 KiB
Python
"""
|
||
把 AI 信号写回数据库。
|
||
跳过纯RT/URL帖子和已有signal的帖子。
|
||
并发执行提速，自动限速避免API超限。
|
||
用法: python scripts/backfill_signals.py [--limit 500] [--overwrite]
|
||
"""
|
||
import asyncio
|
||
import argparse
|
||
import sys
|
||
import os
|
||
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
||
|
||
from app.services.analysis import analyze_post
|
||
from app.database import AsyncSessionLocal
|
||
from app.models import Post
|
||
from sqlalchemy import select
|
||
|
||
CONCURRENCY = 5 # 并发数，避免触发API限速
|
||
|
||
|
||
async def process_one(post: Post, semaphore: asyncio.Semaphore, overwrite: bool) -> str:
|
||
"""分析单条帖子并写回DB，返回状态字符串"""
|
||
if not overwrite and post.signal is not None:
|
||
return "skip"
|
||
|
||
async with semaphore:
|
||
analysis = await analyze_post(post.text)
|
||
await asyncio.sleep(0.1) # 轻度限速
|
||
|
||
async with AsyncSessionLocal() as db:
|
||
result = await db.execute(select(Post).where(Post.id == post.id))
|
||
p = result.scalar_one_or_none()
|
||
if p:
|
||
p.sentiment = analysis["sentiment"]
|
||
p.signal = analysis["signal"]
|
||
p.ai_confidence = analysis["confidence"]
|
||
p.ai_reasoning = analysis["reasoning"]
|
||
p.relevant = analysis["relevant"]
|
||
p.prefilter_reason = analysis.get("prefilter_reason")
|
||
p.analysis_version = analysis.get("analysis_version")
|
||
if analysis["relevant"] and analysis["asset"] and not p.price_impact_asset:
|
||
p.price_impact_asset = analysis["asset"]
|
||
await db.commit()
|
||
|
||
sig = analysis["signal"]
|
||
return f"{sig}:{analysis['confidence']}%"
|
||
|
||
|
||
async def main(limit: int, overwrite: bool):
|
||
async with AsyncSessionLocal() as db:
|
||
result = await db.execute(
|
||
select(Post).order_by(Post.published_at.desc()).limit(limit)
|
||
)
|
||
posts = result.scalars().all()
|
||
|
||
total = len(posts)
|
||
print(f"共 {total} 条帖子，并发={CONCURRENCY}，overwrite={overwrite}\n")
|
||
|
||
semaphore = asyncio.Semaphore(CONCURRENCY)
|
||
tasks = [process_one(p, semaphore, overwrite) for p in posts]
|
||
|
||
done = 0
|
||
buy = short = sell = hold = skipped = 0
|
||
|
||
for coro in asyncio.as_completed(tasks):
|
||
status = await coro
|
||
done += 1
|
||
if status == "skip":
|
||
skipped += 1
|
||
elif status.startswith("buy"): buy += 1
|
||
elif status.startswith("short"): short += 1
|
||
elif status.startswith("sell"): sell += 1
|
||
else: hold += 1
|
||
|
||
if done % 20 == 0 or done == total:
|
||
print(f" 进度 {done}/{total} | BUY={buy} SHORT={short} SELL={sell} HOLD={hold} SKIP={skipped}")
|
||
|
||
print(f"\n✅ 完成: BUY={buy} SHORT={short} SELL={sell} HOLD={hold} SKIP={skipped}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
ap = argparse.ArgumentParser()
|
||
ap.add_argument("--limit", type=int, default=400)
|
||
ap.add_argument("--overwrite", action="store_true", help="重新分析已有signal的帖子")
|
||
args = ap.parse_args()
|
||
asyncio.run(main(args.limit, args.overwrite))
|