{e.detail}
返回看板", status_code=e.status_code) + return templates.TemplateResponse(request=request, name="admin.html", context={"show_nav": True}) + + @router.get("/api/admin/check") + async def api_admin_check(altcoin_session: str = Cookie(default="")): + try: + user = require_admin(altcoin_session) + return {"is_admin": True, "email": user.get("email", "")} + except HTTPException: + return {"is_admin": False} + + @router.get("/api/admin/stats") + async def api_admin_stats(altcoin_session: str = Cookie(default="")): + require_admin(altcoin_session) + return auth_db.get_admin_stats() + + @router.get("/api/admin/users") + async def api_admin_users(search: str = "", offset: int = 0, limit: int = 50, tab: str = "all", altcoin_session: str = Cookie(default="")): + require_admin(altcoin_session) + return auth_db.get_admin_users(search=search, offset=offset, limit=limit, tab=tab) + + @router.get("/api/admin/orders") + async def api_admin_orders(search: str = "", offset: int = 0, limit: int = 50, status: str = "all", altcoin_session: str = Cookie(default="")): + require_admin(altcoin_session) + return auth_db.get_admin_orders(search=search, offset=offset, limit=limit, status=status) + + return router diff --git a/app/web/routes_auth.py b/app/web/routes_auth.py new file mode 100644 index 0000000..17cc3f0 --- /dev/null +++ b/app/web/routes_auth.py @@ -0,0 +1,155 @@ +from fastapi import APIRouter, Cookie, HTTPException, Request +from fastapi.responses import JSONResponse + +from app.db import auth_db +from app.web.shared import ( + ChangePasswordRequest, + CompleteRegistrationRequest, + LoginRequest, + RegisterRequest, + ResendVerificationRequest, + SendCodeRequest, + VerifyEmailRequest, + auth_error, + require_user, +) + +router = APIRouter() + + +@router.post("/api/auth/register") +async def api_auth_register(req: RegisterRequest): + try: + result = auth_db.register_user(req.email, req.password, req.invite_code) + smtp_ready = auth_db.is_smtp_configured() + return { + "ok": True, + "user": {k: v for k, v in result.items() if k not in ("verification_code", "user_id")}, + "dev_verification_code": None if smtp_ready else result.get("verification_code"), + "email_sent": bool(result.get("email_sent")), + "message": "注册成功，请查收邮箱验证码" if smtp_ready else "注册成功，请完成邮箱验证码验证", + } + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/send-code") +async def api_auth_send_code(req: SendCodeRequest): + try: + result = auth_db.send_registration_code(req.email) + smtp_ready = auth_db.is_smtp_configured() + return { + "ok": True, + "email": result["email"], + "dev_verification_code": None if smtp_ready else result.get("verification_code"), + "email_sent": bool(result.get("email_sent")), + "message": "验证码已发送，请查收邮箱" if smtp_ready else "验证码已生成", + } + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/complete-registration") +async def api_auth_complete_registration(req: CompleteRegistrationRequest): + try: + user = auth_db.complete_registration(req.email, req.code, req.password, req.invite_code) + return {"ok": True, "user": user, "message": "注册成功，请登录"} + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/verify-email") +async def api_auth_verify_email(req: VerifyEmailRequest): + try: + user = auth_db.verify_email(req.email, req.code) + return {"ok": True, "user": user, "message": "邮箱验证成功"} + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/resend-verification") +async def api_auth_resend_verification(req: ResendVerificationRequest): + try: + result = auth_db.resend_verification_code(req.email) + smtp_ready = auth_db.is_smtp_configured() + return { + "ok": True, + "email": result["email"], + "dev_verification_code": None if smtp_ready else result.get("verification_code"), + "email_sent": bool(result.get("email_sent")), + "message": "验证码已重新发送，请查收邮箱" if smtp_ready else "验证码已重新生成", + } + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/login") +async def api_auth_login(req: LoginRequest, request: Request = None): + try: + session = auth_db.login_user(req.email, req.password) + auth_db.log_user_activity(session["user"]["id"], "login", "auth", ip=request.client.host if request.client else "") + sub = auth_db.get_current_subscription(session["user"]["id"]) + next_path = "/app" if sub else "/subscription?welcome=1" + resp = JSONResponse({ + "ok": True, + "user": session["user"], + "expires_at": session["expires_at"], + "subscription": sub, + "subscription_active": bool(sub), + "next": next_path, + }) + resp.set_cookie("altcoin_session", session["token"], httponly=True, samesite="lax", max_age=30 * 24 * 3600) + return resp + except auth_db.AuthError as exc: + auth_error(exc, status_code=400) + + +@router.get("/api/auth/me") +async def api_auth_me(altcoin_session: str = Cookie(default="")): + user = require_user(altcoin_session) + sub = auth_db.get_current_subscription(user["id"]) + return {"ok": True, "user": user, "subscription": sub, "subscription_active": bool(sub)} + + +@router.post("/api/auth/change-password") +async def api_auth_change_password(req: ChangePasswordRequest, altcoin_session: str = Cookie(default="")): + user = require_user(altcoin_session) + try: + return auth_db.change_password(user["id"], req.old_password, req.new_password) + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.post("/api/auth/logout") +async def api_auth_logout(altcoin_session: str = Cookie(default="")): + auth_db.logout_user(altcoin_session) + resp = JSONResponse({"ok": True, "message": "已退出登录"}) + resp.delete_cookie("altcoin_session") + return resp + + +@router.post("/api/subscriptions/free-trial") +async def api_subscription_free_trial(altcoin_session: str = Cookie(default="")): + user = require_user(altcoin_session) + try: + sub = auth_db.claim_free_trial(user["id"]) + return {"ok": True, "subscription": sub, "message": "已开通新用户免费体验1个月"} + except auth_db.AuthError as exc: + auth_error(exc) + + +@router.get("/api/subscription/plans") +async def api_subscription_plans(): + auth_db.init_auth_db() + conn = auth_db.get_conn() + rows = conn.execute("SELECT * FROM subscription_plan ORDER BY sort_order ASC").fetchall() + conn.close() + return [dict(r) for r in rows] + + +@router.get("/api/referral/stats") +async def api_referral_stats(altcoin_session: str = Cookie(default="")): + user = auth_db.get_user_by_session_token(altcoin_session) + if not user: + raise HTTPException(status_code=401, detail="请先登录") + return auth_db.get_referral_stats(user["id"]) diff --git a/app/web/routes_content.py b/app/web/routes_content.py new file mode 100644 index 0000000..5b794d3 --- /dev/null +++ b/app/web/routes_content.py @@ -0,0 +1,294 @@ +import json +import os +import sqlite3 +from datetime import datetime, timezone +from pathlib import Path + +from fastapi import APIRouter, Cookie +from fastapi.responses import JSONResponse + +from app.web.shared import require_api_user_with_subscription + + +def build_router(repo_root: Path): + router = APIRouter() + + @router.get("/api/sentiment") + async def api_sentiment(hours: int = 6, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + db = os.getenv("ALPHAX_DB_PATH", str(repo_root / "data" / "altcoin_monitor.db")) + conn = sqlite3.connect(db) + conn.row_factory = sqlite3.Row + + active_recs = conn.execute("SELECT DISTINCT symbol FROM recommendation WHERE status='active'").fetchall() + active_symbols = {r["symbol"].split("/")[0].upper() for r in active_recs} + + recent_screened = conn.execute( + """ + SELECT DISTINCT symbol FROM screening_log + WHERE scan_time >= datetime('now', '-' || ? || ' hours') + """, + (hours,), + ).fetchall() + screened_bases = {r["symbol"].split("/")[0].upper() for r in recent_screened} + + events = [] + now_utc = datetime.now(timezone.utc) + + def _parse_event_time(value): + if not value: + return None + text = str(value).strip() + for fmt in ("%a, %d %b %Y %H:%M:%S %Z", "%a, %d %b %Y %H:%M:%S GMT"): + try: + return datetime.strptime(text, fmt).replace(tzinfo=timezone.utc) + except Exception: + pass + try: + dt = datetime.fromisoformat(text.replace("Z", "+00:00")) + if dt.tzinfo is None: + dt = dt.replace(tzinfo=timezone.utc) + return dt.astimezone(timezone.utc) + except Exception: + return None + + def _is_fresh_news(value, max_hours): + dt = _parse_event_time(value) + if not dt: + return False + age_hours = (now_utc - dt).total_seconds() / 3600 + return 0 <= age_hours <= max_hours + + valuable_news_keywords = [ + "listing", "listed", "launch", "launchpool", "megadrop", "airdrop", "mainnet", "upgrade", + "partnership", "integrat", "acquisition", "merge", "buyback", "burn", "token burn", "funding", + "raises", "investment", "sec", "etf", "approval", "lawsuit", "settlement", "hack", "exploit", + "delist", "suspend", "roadmap", "migration", "上线", "上币", "合约", "空投", "主网", "升级", "合作", + "收购", "回购", "销毁", "融资", "获投", "监管", "批准", "黑客", "漏洞", "下架", "暂停", + ] + low_value_news_keywords = [ + "price prediction", "price today", "live price", "marketcap and chart", "what is", "how to buy", + "good investment", "forecast", "prediction 2026", "prediction 2027", "prediction 2030", + "technical analysis", "is it time to buy", "价格预测", "今日价格", "实时价格", "怎么买", "是什么币", + ] + + def _is_valuable_news_title(title): + text = (title or "").lower() + if not text: + return False + if any(k in text for k in low_value_news_keywords): + return False + return any(k in text for k in valuable_news_keywords) + + try: + event_rows = conn.execute( + """ + SELECT source, symbol, title, url, published_at, detected_at, importance, + event_type, decision, tech_score, rec_id, pushed + FROM event_news + WHERE detected_at >= datetime('now', '-' || ? || ' hours') + ORDER BY datetime(published_at) DESC, id DESC + LIMIT 80 + """, + (hours,), + ).fetchall() + for r in event_rows: + base = (r["symbol"] or "").split("/")[0].upper() + source = r["source"] or "event" + event_type = r["event_type"] or "event" + if event_type == "market_heat": + continue + events.append({ + "source": source, + "source_label": "Binance公告" if "binance" in source else "CoinGecko热度" if "coingecko" in source else source, + "event_type": event_type, + "importance": r["importance"] or "B", + "title": r["title"] or "", + "url": r["url"] or "", + "published_at": r["published_at"], + "detected_at": r["detected_at"], + "related_symbol": r["symbol"], + "related_base": base, + "related_name": "", + "decision": r["decision"] or "", + "tech_score": r["tech_score"] or 0, + "rec_id": r["rec_id"] or 0, + "pushed": bool(r["pushed"]), + "in_active": base in active_symbols, + "in_screened": base in screened_bases, + "price_usd": 0, + "change_24h_pct": 0, + "market_cap_rank": 0, + "trend_rank": None, + }) + except Exception: + pass + + rows = conn.execute( + """ + SELECT symbol, name, trend_rank, trend_score, market_cap_rank, detected_at, extra_json + FROM sentiment_events + WHERE detected_at = (SELECT MAX(detected_at) FROM sentiment_events WHERE source='coingecko') + ORDER BY trend_rank + """ + ).fetchall() + for r in rows: + raw_extra = r["extra_json"] + if not raw_extra or not isinstance(raw_extra, str) or not raw_extra.strip(): + extra = {} + else: + try: + extra = json.loads(raw_extra) + except Exception: + extra = {} + base = (r["symbol"] or "").upper() + name = r["name"] or base + price_usd = extra.get("price_usd", 0) or 0 + change_24h_pct = extra.get("change_24h_pct", 0) or 0 + news_items = extra.get("news", []) or [] + + for n in news_items[:3]: + published = n.get("published") or "" + if not _is_fresh_news(published, hours): + continue + title = n.get("title") or f"{name} 相关新闻" + if not _is_valuable_news_title(title): + continue + events.append({ + "source": n.get("source") or "news", + "source_label": n.get("source") or "新闻", + "event_type": "news", + "importance": "B", + "title": title, + "url": n.get("url") or "", + "published_at": published, + "detected_at": r["detected_at"], + "related_symbol": f"{base}/USDT", + "related_base": base, + "related_name": name, + "decision": "", + "tech_score": 0, + "rec_id": 0, + "pushed": False, + "in_active": base in active_symbols, + "in_screened": base in screened_bases, + "price_usd": price_usd, + "change_24h_pct": change_24h_pct, + "market_cap_rank": r["market_cap_rank"], + "trend_rank": r["trend_rank"], + }) + + conn.close() + + deduped = [] + seen = set() + for e in events: + key = ((e.get("title") or "").strip().lower(), e.get("related_base"), e.get("source")) + if key in seen: + continue + seen.add(key) + if e.get("in_active"): + e["relation_tag"] = "持仓/活跃推荐" + elif e.get("in_screened"): + e["relation_tag"] = "系统筛选中" + else: + e["relation_tag"] = "关联币种" + deduped.append(e) + + deduped.sort(key=lambda item: (item.get("published_at") or item.get("detected_at") or "", {"RISK": 5, "S": 4, "A": 3, "B": 2, "C": 1}.get(item.get("importance"), 0)), reverse=True) + check_time = deduped[0]["detected_at"] if deduped else None + return { + "check_time": check_time, + "total_events": len(deduped), + "overlap_active": sum(1 for e in deduped if e["in_active"]), + "overlap_screened": sum(1 for e in deduped if e["in_screened"]), + "events": deduped[:80], + "trending": [], + "total_trending": 0, + } + + @router.get("/api/kline") + async def api_kline(symbol: str, interval: str = "1d", limit: int = 60, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + import requests as req + + try: + clean = symbol.replace("/", "") + r = req.get( + "https://api.binance.com/api/v3/klines", + params={"symbol": clean, "interval": interval, "limit": limit}, + timeout=10, + ) + if r.status_code != 200: + return JSONResponse({"error": f"Binance {r.status_code}"}, status_code=502) + data = r.json() + candles = [ + {"time": d[0], "open": float(d[1]), "high": float(d[2]), "low": float(d[3]), "close": float(d[4]), "volume": float(d[5])} + for d in data + ] + return {"symbol": symbol, "interval": interval, "candles": candles} + except Exception as e: + return JSONResponse({"error": str(e)}, status_code=500) + + @router.get("/api/newsfeed") + async def api_newsfeed(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + import requests as req + import xml.etree.ElementTree as ET + from email.utils import parsedate_to_datetime + + result = {"fear_greed": None, "trending": [], "news": []} + now = datetime.now(timezone.utc) + try: + r = req.get("https://api.alternative.me/fng/?limit=1", timeout=8) + if r.status_code == 200: + d0 = r.json().get("data", [{}])[0] + result["fear_greed"] = {"value": int(d0.get("value", 50)), "classification": d0.get("value_classification", "")} + except Exception: + pass + + try: + r = req.get("https://api.coingecko.com/api/v3/search/trending", timeout=10) + if r.status_code == 200: + for c in r.json().get("coins", [])[:7]: + item = c.get("item", {}) + result["trending"].append({ + "name": item.get("name", ""), + "symbol": item.get("symbol", ""), + "market_cap_rank": item.get("market_cap_rank"), + "thumb": item.get("thumb", ""), + }) + except Exception: + pass + + def fetch_google_news(query, hl, gl, ceid, label): + items = [] + try: + url = f"https://news.google.com/rss/search?q={req.utils.quote(query)}&hl={hl}&gl={gl}&ceid={ceid}" + r = req.get(url, timeout=12, headers={"User-Agent": "Mozilla/5.0"}) + if r.status_code != 200: + return items + root = ET.fromstring(r.text) + for el in root.findall(".//item")[:15]: + pub_str = el.findtext("pubDate", "") + dt = parsedate_to_datetime(pub_str) if pub_str else None + age_h = round((now - dt).total_seconds() / 3600, 1) if dt else None + if age_h is not None and age_h > 48: + continue + items.append({ + "title": (el.findtext("title", "") or "")[:120], + "url": el.findtext("link", "") or "", + "source": (el.findtext("source", "") or "")[:30], + "age_hours": age_h, + "lang": label, + }) + except Exception: + pass + return items + + en_news = fetch_google_news("cryptocurrency OR bitcoin OR ethereum OR defi OR altcoin when:24h", "en-US", "US", "US:en", "en") + cn_news = fetch_google_news("加密货币 OR 比特币 OR 以太坊 OR DeFi OR Web3 when:24h", "zh-CN", "CN", "CN:zh-Hans", "cn") + result["news"] = sorted(en_news + cn_news, key=lambda x: x.get("age_hours") or 999)[:30] + return result + + return router diff --git a/app/web/routes_pages.py b/app/web/routes_pages.py new file mode 100644 index 0000000..cc995bd --- /dev/null +++ b/app/web/routes_pages.py @@ -0,0 +1,98 @@ +from pathlib import Path + +from fastapi import APIRouter, Cookie, Request +from fastapi.responses import HTMLResponse + +from app.db import auth_db +from app.web.shared import require_page_user + + +def build_router(templates, repo_root: Path, stock_report_template: str): + router = APIRouter() + + def render_page(template_name: str, request: Request, **kwargs): + try: + user = auth_db.get_user_by_session_token(request.cookies.get("altcoin_session", "")) + if user: + auth_db.log_user_activity( + user["id"], + "page_view", + template_name.replace(".html", ""), + ip=request.client.host if request.client else "", + ) + except Exception: + pass + return templates.TemplateResponse(request=request, name=template_name, context={"show_nav": True, **kwargs}) + + @router.get("/", response_class=HTMLResponse) + async def index(): + with open(repo_root / "static" / "index.html", "r", encoding="utf-8") as f: + return HTMLResponse(content=f.read()) + + @router.get("/auth", response_class=HTMLResponse) + async def auth_page(): + with open(repo_root / "static" / "auth.html", "r", encoding="utf-8") as f: + return HTMLResponse(content=f.read()) + + @router.get("/watchlist", response_class=HTMLResponse) + async def watchlist_page(request: Request): + user, redirect = require_page_user(request) + if redirect: + return redirect + return render_page("watchlist.html", request) + + @router.get("/strategy", response_class=HTMLResponse) + async def strategy_page(request: Request): + user, redirect = require_page_user(request) + if redirect: + return redirect + return render_page("strategy.html", request) + + @router.get("/subscription", response_class=HTMLResponse) + async def subscription_page(request: Request): + user, redirect = require_page_user(request, require_subscription=False) + if redirect: + return redirect + return render_page("subscription.html", request) + + @router.get("/referral", response_class=HTMLResponse) + async def referral_page(request: Request, altcoin_session: str = Cookie(default="")): + user, redirect = require_page_user(request) + if redirect: + return redirect + return render_page("referral.html", request) + + @router.get("/app", response_class=HTMLResponse) + async def app_page(altcoin_session: str = Cookie(default=""), request: Request = None): + user, redirect = require_page_user(request) + if redirect: + return redirect + try: + auth_db.log_user_activity(user["id"], "page_view", "app", ip=request.client.host if request and request.client else "") + except Exception: + pass + resp = templates.TemplateResponse(request=request, name="app.html", context={"show_nav": True}) + resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" + resp.headers["Pragma"] = "no-cache" + resp.headers["Expires"] = "0" + return resp + + @router.get("/sentiment", response_class=HTMLResponse) + async def sentiment_page(request: Request): + user, redirect = require_page_user(request) + if redirect: + return redirect + return render_page("sentiment.html", request) + + @router.get("/iteration", response_class=HTMLResponse) + async def iteration_page(request: Request): + user, redirect = require_page_user(request) + if redirect: + return redirect + return render_page("iteration.html", request) + + @router.get("/stock-report", response_class=HTMLResponse) + async def stock_report_page(): + return HTMLResponse(content=stock_report_template) + + return router diff --git a/app/web/routes_recommendations.py b/app/web/routes_recommendations.py new file mode 100644 index 0000000..1c311dc --- /dev/null +++ b/app/web/routes_recommendations.py @@ -0,0 +1,141 @@ +from fastapi import APIRouter, Cookie + +from app.db import auth_db +from app.db.analytics import ( + get_all_recommendations, + get_cron_run_logs, + get_cron_run_summary, + get_review_stats, + get_screening_history, + get_stats, +) +from app.db.recommendation_queries import get_active_recommendations, get_active_recommendations_deduped +from app.config.config_loader import get_signal_weights +from app.web.shared import ( + ObservationRequest, + PushRulesRequest, + WatchlistRequest, + require_api_user_with_subscription, +) + +router = APIRouter() + + +@router.get("/api/stats") +async def api_stats(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_stats() + + +@router.get("/api/recommendations") +async def api_recommendations( + limit: int = 50, + offset: int = 0, + decision_only: bool = False, + version: str = "", + paged: bool = False, + compact: bool = False, + altcoin_session: str = Cookie(default=""), +): + require_api_user_with_subscription(altcoin_session) + return get_all_recommendations(limit, decision_only=decision_only, version=version, offset=offset, with_meta=(paged or compact)) + + +@router.get("/api/recommendations/active") +async def api_recommendations_active( + dedup: bool = True, + actionable_only: bool = True, + version: str = "", + hours: float = 0, + limit: int = 0, + offset: int = 0, + paged: bool = False, + compact: bool = False, + altcoin_session: str = Cookie(default=""), +): + require_api_user_with_subscription(altcoin_session) + if dedup: + return get_active_recommendations_deduped( + actionable_only=actionable_only, + version=version, + hours=hours, + limit=limit, + offset=offset, + with_meta=(paged or compact), + ) + return get_active_recommendations(actionable_only=actionable_only) + + +@router.get("/api/personalization") +async def api_personalization(altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + return { + "watchlist": auth_db.get_watchlist_symbols(user["id"]), + "observations": auth_db.get_saved_observations(user["id"]), + "push_rules": auth_db.get_push_rules(user["id"]), + } + + +@router.post("/api/watchlist") +async def api_add_watchlist(req: WatchlistRequest, altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + auth_db.add_watchlist_symbol(user["id"], req.symbol) + return {"ok": True, "watchlist": auth_db.get_watchlist_symbols(user["id"])} + + +@router.delete("/api/watchlist/{symbol}") +async def api_remove_watchlist(symbol: str, altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + auth_db.remove_watchlist_symbol(user["id"], symbol) + return {"ok": True, "watchlist": auth_db.get_watchlist_symbols(user["id"])} + + +@router.post("/api/observations") +async def api_save_observation(req: ObservationRequest, altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + auth_db.save_observation(user["id"], req.rec_id, req.note) + return {"ok": True, "observations": auth_db.get_saved_observations(user["id"])} + + +@router.delete("/api/observations/{rec_id}") +async def api_remove_observation(rec_id: int, altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + auth_db.remove_observation(user["id"], rec_id) + return {"ok": True, "observations": auth_db.get_saved_observations(user["id"])} + + +@router.post("/api/push-rules") +async def api_update_push_rules(req: PushRulesRequest, altcoin_session: str = Cookie(default="")): + user = require_api_user_with_subscription(altcoin_session) + rules = auth_db.update_push_rules(user["id"], req.dict()) + return {"ok": True, "push_rules": rules} + + +@router.get("/api/screening") +async def api_screening(hours: int = 24, limit: int = 100, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_screening_history(hours, limit) + + +@router.get("/api/review") +async def api_review(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_review_stats() + + +@router.get("/api/weights") +async def api_weights(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_signal_weights() + + +@router.get("/api/cron") +async def api_cron(limit: int = 50, job_name: str = "", altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_cron_run_logs(limit=limit, job_name=job_name or None) + + +@router.get("/api/cron/summary") +async def api_cron_summary(hours: int = 24, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_cron_run_summary(hours=hours) diff --git a/app/web/routes_strategy.py b/app/web/routes_strategy.py new file mode 100644 index 0000000..db5d04f --- /dev/null +++ b/app/web/routes_strategy.py @@ -0,0 +1,123 @@ +from fastapi import APIRouter, Cookie + +from app.config.config_loader import get_meta +from app.db import auth_db +from app.db.review_queries import ( + backfill_strategy_failure_patterns, + dry_run_strategy_candidate_performance, + generate_candidates_from_review_history, + get_strategy_failure_patterns, + get_strategy_insights, + get_strategy_iteration_dashboard, + get_strategy_rule_candidates, + refresh_strategy_candidate_performance, +) +from app.db.schema import get_conn +from app.db.altcoin_db import _derive_execution_fields +from app.web.shared import require_api_user_with_subscription + +router = APIRouter() + + +@router.get("/api/versions") +async def api_versions(view: str = "active", altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + conn = get_conn() + rows = conn.execute( + """ + SELECT r.* FROM recommendation r + JOIN ( + SELECT symbol, strategy_version, MAX(id) AS max_id + FROM recommendation + WHERE status='active' AND strategy_version IS NOT NULL AND strategy_version != '' + GROUP BY symbol, strategy_version + ) latest ON latest.max_id = r.id + ORDER BY r.strategy_version DESC, r.rec_time DESC + """ + ).fetchall() + conn.close() + counts = {} + for row in rows: + item = dict(row) + _derive_execution_fields(item) + version = str(item.get("strategy_version") or "").strip() + if not version: + continue + status = item.get("execution_status") or "observe" + if view == "active" and status not in ("buy_now", "wait_pullback"): + continue + if view == "watch" and status != "observe": + continue + counts[version] = counts.get(version, 0) + 1 + + versions = [{"version": version, "count": count} for version, count in counts.items()] + current_version = str(get_meta().get("strategy_version") or "").strip() + if current_version and current_version not in counts: + versions.append({"version": current_version, "count": 0}) + + def _version_key(v): + try: + return tuple(int(p) for p in v["version"].lstrip("v").split(".")) + except Exception: + return (0,) + + versions.sort(key=_version_key, reverse=True) + return versions + + +@router.get("/api/strategy/insights") +async def api_strategy_insights(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_strategy_insights() + + +@router.get("/api/strategy/lifecycle") +async def api_strategy_lifecycle(days: int = 30, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_strategy_iteration_dashboard(days=days) + + +@router.get("/api/iterations") +async def api_iterations(limit: int = 30, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + conn = get_conn() + rows = conn.execute("SELECT * FROM strategy_iteration_log ORDER BY id DESC LIMIT ?", (limit,)).fetchall() + conn.close() + return [dict(r) for r in rows] + + +@router.get("/api/strategy/candidates") +async def api_strategy_candidates(limit: int = 50, status: str = "", altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_strategy_rule_candidates(limit=limit, status=status or None) + + +@router.get("/api/strategy/failures") +async def api_strategy_failures(limit: int = 50, altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return get_strategy_failure_patterns(limit=limit) + + +@router.post("/api/strategy/candidates/refresh") +async def api_strategy_candidates_refresh(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return {"updated": refresh_strategy_candidate_performance()} + + +@router.get("/api/strategy/candidates/dry-run") +async def api_strategy_candidates_dry_run(altcoin_session: str = Cookie(default="")): + require_api_user_with_subscription(altcoin_session) + return dry_run_strategy_candidate_performance() + + +@router.post("/api/strategy/failures/backfill") +async def api_strategy_failures_backfill(dry_run: bool = False): + return backfill_strategy_failure_patterns(dry_run=dry_run) + + +@router.post("/api/strategy/candidates/generate-history") +async def api_strategy_candidates_generate_history(dry_run: bool = False): + result = generate_candidates_from_review_history(dry_run=dry_run) + if not dry_run: + result["refreshed"] = refresh_strategy_candidate_performance() + return result diff --git a/app/web/shared.py b/app/web/shared.py new file mode 100644 index 0000000..ba326a8 --- /dev/null +++ b/app/web/shared.py @@ -0,0 +1,156 @@ +"""Shared web-layer request models and auth helpers.""" + +from contextvars import ContextVar + +from fastapi import Cookie, HTTPException, Request +from fastapi.responses import RedirectResponse +from pydantic import BaseModel + +from app.db import auth_db + +current_request = ContextVar("current_request", default=None) + + +class RegisterRequest(BaseModel): + email: str + password: str + invite_code: str = "" + + +class VerifyEmailRequest(BaseModel): + email: str + code: str + + +class LoginRequest(BaseModel): + email: str + password: str + + +class ResendVerificationRequest(BaseModel): + email: str + + +class SendCodeRequest(BaseModel): + email: str + + +class CompleteRegistrationRequest(BaseModel): + email: str + code: str + password: str + invite_code: str = "" + + +class ChangePasswordRequest(BaseModel): + old_password: str + new_password: str + + +class WatchlistRequest(BaseModel): + symbol: str + + +class ObservationRequest(BaseModel): + rec_id: int + note: str = "" + + +class PushRulesRequest(BaseModel): + watchlist_only: bool = False + min_score: int = 0 + min_rr: float = 0 + push_buy_now: bool = True + push_wait_pullback: bool = True + push_observe: bool = False + quiet_start: str = "" + quiet_end: str = "" + + +def auth_error(exc: Exception, status_code: int = 400): + raise HTTPException(status_code=status_code, detail=str(exc)) + + +def is_local_request(request: Request = None) -> bool: + request = request or current_request.get() + if not request or not request.client: + return False + host = (request.client.host or "").split(":")[0] + return host in ("127.0.0.1", "localhost", "::1", "testclient") + + +def local_debug_user(): + return {"id": 0, "email": "local@alphax.dev", "is_admin": True, "local_debug": True} + + +def require_user(altcoin_session: str = ""): + user = auth_db.get_user_by_session_token(altcoin_session) + if user: + return user + if is_local_request(): + return local_debug_user() + if not user: + raise HTTPException(status_code=401, detail="请先登录") + return user + + +def require_active_subscription(altcoin_session: str = ""): + user = require_user(altcoin_session) + if user.get("local_debug"): + return user, {"plan_name": "本地调试", "local_debug": True} + sub = auth_db.get_current_subscription(user["id"]) + if not sub: + raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅") + return user, sub + + +def require_admin(altcoin_session: str = ""): + user = require_user(altcoin_session) + if user.get("local_debug"): + return user + if not auth_db.is_user_admin(user["id"]): + raise HTTPException(status_code=403, detail="需要管理员权限") + return user + + +def login_redirect(): + return RedirectResponse(url="/auth?tab=login", status_code=302) + + +def subscription_redirect(): + return RedirectResponse(url="/subscription?expired=1", status_code=302) + + +def has_active_subscription(user) -> bool: + if is_local_request(): + return True + if not user: + return False + try: + if auth_db.is_user_admin(user["id"]): + return True + except Exception: + pass + return bool(auth_db.get_current_subscription(user["id"])) + + +def require_page_user(request: Request, require_subscription: bool = True): + user = auth_db.get_user_by_session_token(request.cookies.get("altcoin_session", "")) + if user: + if require_subscription and not has_active_subscription(user): + return user, subscription_redirect() + return user, None + if is_local_request(request): + return local_debug_user(), None + if not user: + return None, login_redirect() + return user, None + + +def require_api_user_with_subscription(altcoin_session: str = Cookie(default="")): + user = require_user(altcoin_session) + if user.get("local_debug"): + return user + if not has_active_subscription(user): + raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅") + return user diff --git a/app/web/web_server.py b/app/web/web_server.py index 1487d38..710944a 100644 --- a/app/web/web_server.py +++ b/app/web/web_server.py @@ -1,4255 +1,52 @@ """ -山寨币监控网站 — FastAPI后端 v11（纯前瞻信号版） -新增：复盘tab + 信号权重API +FastAPI application assembly for the AlphaX web surface. """ -import sys -import os -import json -import sqlite3 -from datetime import datetime, timezone -from contextvars import ContextVar +from contextlib import asynccontextmanager from pathlib import Path -from fastapi import FastAPI, HTTPException, Cookie, Request -from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse -from fastapi.templating import Jinja2Templates -from pydantic import BaseModel +from fastapi import FastAPI, Request +from fastapi.templating import Jinja2Templates + +from app.db.schema import init_db from app.db import auth_db -from app.db.altcoin_db import ( - init_db, get_active_recommendations, get_active_recommendations_deduped, get_all_recommendations, - get_screening_history, get_stats, get_review_stats, get_cron_run_logs, get_cron_run_summary, - get_conn, _derive_execution_fields, get_strategy_insights, get_strategy_rule_candidates, - get_strategy_failure_patterns, get_strategy_iteration_dashboard, refresh_strategy_candidate_performance, - dry_run_strategy_candidate_performance, backfill_strategy_failure_patterns, - generate_candidates_from_review_history, -) -from app.config.config_loader import get_signal_weights, get_meta +from app.db.analytics import get_all_recommendations, get_cron_run_logs, get_cron_run_summary, get_review_stats, get_stats +from app.db.recommendation_queries import get_active_recommendations, get_active_recommendations_deduped +from app.web.routes_admin import build_router as build_admin_router +from app.web.routes_auth import router as auth_router +from app.web.routes_content import build_router as build_content_router +from app.web.routes_pages import build_router as build_pages_router +from app.web.routes_recommendations import router as recommendations_router +from app.web.routes_strategy import router as strategy_router +from app.web.shared import current_request +from app.web.shared import require_active_subscription as _require_active_subscription REPO_ROOT = Path(__file__).resolve().parents[2] +STOCK_REPORT_TEMPLATE = (REPO_ROOT / "templates" / "stock_report_template.html").read_text(encoding="utf-8") +HTML_PAGE = (REPO_ROOT / "static" / "app.html").read_text(encoding="utf-8") -app = FastAPI(title="山寨币爆发监控 v11") + +@asynccontextmanager +async def lifespan(app: FastAPI): + init_db() + yield + + +app = FastAPI(title="山寨币爆发监控 v11", lifespan=lifespan) templates = Jinja2Templates(directory=str(REPO_ROOT / "static")) -_current_request = ContextVar("current_request", default=None) + +app.include_router(auth_router) +app.include_router(recommendations_router) +app.include_router(strategy_router) +app.include_router(build_admin_router(templates)) +app.include_router(build_content_router(REPO_ROOT)) +app.include_router(build_pages_router(templates, REPO_ROOT, STOCK_REPORT_TEMPLATE)) @app.middleware("http") async def bind_current_request(request: Request, call_next): - token = _current_request.set(request) + token = current_request.set(request) try: return await call_next(request) finally: - _current_request.reset(token) - - -def _is_local_request(request: Request = None) -> bool: - """仅本机调试访问免登录；公网/域名访问仍走完整认证与订阅闸门。""" - request = request or _current_request.get() - if not request or not request.client: - return False - host = (request.client.host or "").split(":")[0] - return host in ("127.0.0.1", "localhost", "::1") - - -def _local_debug_user(): - return {"id": 0, "email": "local@alphax.dev", "is_admin": True, "local_debug": True} - - -class RegisterRequest(BaseModel): - email: str - password: str - invite_code: str = "" - - -class VerifyEmailRequest(BaseModel): - email: str - code: str - - -class LoginRequest(BaseModel): - email: str - password: str - - -class ResendVerificationRequest(BaseModel): - email: str - - -class SendCodeRequest(BaseModel): - email: str - - -class CompleteRegistrationRequest(BaseModel): - email: str - code: str - password: str - invite_code: str = "" - - -class ChangePasswordRequest(BaseModel): - old_password: str - new_password: str - - -class WatchlistRequest(BaseModel): - symbol: str - - -class ObservationRequest(BaseModel): - rec_id: int - note: str = "" - - -class PushRulesRequest(BaseModel): - watchlist_only: bool = False - min_score: int = 0 - min_rr: float = 0 - push_buy_now: bool = True - push_wait_pullback: bool = True - push_observe: bool = False - quiet_start: str = "" - quiet_end: str = "" - - -def _auth_error(exc: Exception, status_code: int = 400): - raise HTTPException(status_code=status_code, detail=str(exc)) - - -def _require_user(altcoin_session: str = ""): - if _is_local_request(): - return _local_debug_user() - user = auth_db.get_user_by_session_token(altcoin_session) - if not user: - raise HTTPException(status_code=401, detail="请先登录") - return user - - -def _require_active_subscription(altcoin_session: str = ""): - if _is_local_request(): - return _local_debug_user(), {"plan_name": "本地调试", "local_debug": True} - user = _require_user(altcoin_session) - sub = auth_db.get_current_subscription(user["id"]) - if not sub: - raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅") - return user, sub - - -def _require_admin(altcoin_session: str = ""): - """v1.7.8: 管理员权限校验""" - if _is_local_request(): - return _local_debug_user() - user = _require_user(altcoin_session) - if not auth_db.is_user_admin(user["id"]): - raise HTTPException(status_code=403, detail="需要管理员权限") - return user - - -def _login_redirect(): - return RedirectResponse(url="/auth?tab=login", status_code=302) - - -def _subscription_redirect(): - return RedirectResponse(url="/subscription?expired=1", status_code=302) - - -def _has_active_subscription(user) -> bool: - """页面访问用订阅校验：管理员放行；普通用户必须有未过期订阅。""" - if _is_local_request(): - return True - if not user: - return False - try: - if auth_db.is_user_admin(user["id"]): - return True - except Exception: - pass - return bool(auth_db.get_current_subscription(user["id"])) - - -def _require_page_user(request: Request, require_subscription: bool = True): - if _is_local_request(request): - return _local_debug_user(), None - user = auth_db.get_user_by_session_token(request.cookies.get("altcoin_session", "")) - if not user: - return None, _login_redirect() - if require_subscription and not _has_active_subscription(user): - return user, _subscription_redirect() - return user, None - - -def _require_api_user_with_subscription(altcoin_session: str = ""): - if _is_local_request(): - return _local_debug_user() - user = _require_user(altcoin_session) - if not _has_active_subscription(user): - raise HTTPException(status_code=402, detail="订阅已过期或未开通，请先开通订阅") - return user - -@app.post("/api/auth/register") -async def api_auth_register(req: RegisterRequest): - try: - result = auth_db.register_user(req.email, req.password, req.invite_code) - # SMTP 已配置时验证码只通过邮件发送；未配置时返回 dev_verification_code 便于本地测试。 - smtp_ready = auth_db.is_smtp_configured() - return { - "ok": True, - "user": {k: v for k, v in result.items() if k not in ("verification_code", "user_id")}, - "dev_verification_code": None if smtp_ready else result.get("verification_code"), - "email_sent": bool(result.get("email_sent")), - "message": "注册成功，请查收邮箱验证码" if smtp_ready else "注册成功，请完成邮箱验证码验证", - } - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/send-code") -async def api_auth_send_code(req: SendCodeRequest): - """注册步骤1：仅发送邮箱验证码，不要求密码。""" - try: - result = auth_db.send_registration_code(req.email) - smtp_ready = auth_db.is_smtp_configured() - return { - "ok": True, - "email": result["email"], - "dev_verification_code": None if smtp_ready else result.get("verification_code"), - "email_sent": bool(result.get("email_sent")), - "message": "验证码已发送，请查收邮箱" if smtp_ready else "验证码已生成", - } - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/complete-registration") -async def api_auth_complete_registration(req: CompleteRegistrationRequest): - """注册步骤2：验证验证码 + 设置密码 + 完成注册。""" - try: - user = auth_db.complete_registration(req.email, req.code, req.password, req.invite_code) - return {"ok": True, "user": user, "message": "注册成功，请登录"} - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/verify-email") -async def api_auth_verify_email(req: VerifyEmailRequest): - try: - user = auth_db.verify_email(req.email, req.code) - return {"ok": True, "user": user, "message": "邮箱验证成功"} - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/resend-verification") -async def api_auth_resend_verification(req: ResendVerificationRequest): - try: - result = auth_db.resend_verification_code(req.email) - smtp_ready = auth_db.is_smtp_configured() - return { - "ok": True, - "email": result["email"], - "dev_verification_code": None if smtp_ready else result.get("verification_code"), - "email_sent": bool(result.get("email_sent")), - "message": "验证码已重新发送，请查收邮箱" if smtp_ready else "验证码已重新生成", - } - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/login") -async def api_auth_login(req: LoginRequest, request: Request = None): - try: - session = auth_db.login_user(req.email, req.password) - auth_db.log_user_activity(session["user"]["id"], "login", "auth", - ip=request.client.host if request.client else "") - sub = auth_db.get_current_subscription(session["user"]["id"]) - next_path = "/app" if sub else "/subscription?welcome=1" - resp = JSONResponse({ - "ok": True, - "user": session["user"], - "expires_at": session["expires_at"], - "subscription": sub, - "subscription_active": bool(sub), - "next": next_path, - }) - resp.set_cookie( - "altcoin_session", - session["token"], - httponly=True, - samesite="lax", - max_age=30 * 24 * 3600, - ) - return resp - except auth_db.AuthError as exc: - _auth_error(exc, status_code=400) - - -@app.get("/api/auth/me") -async def api_auth_me(altcoin_session: str = Cookie(default="")): - user = _require_user(altcoin_session) - sub = auth_db.get_current_subscription(user["id"]) - return {"ok": True, "user": user, "subscription": sub, "subscription_active": bool(sub)} - - -@app.post("/api/auth/change-password") -async def api_auth_change_password(req: ChangePasswordRequest, altcoin_session: str = Cookie(default="")): - user = _require_user(altcoin_session) - try: - result = auth_db.change_password(user["id"], req.old_password, req.new_password) - return result - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.post("/api/auth/logout") -async def api_auth_logout(altcoin_session: str = Cookie(default="")): - auth_db.logout_user(altcoin_session) - resp = JSONResponse({"ok": True, "message": "已退出登录"}) - resp.delete_cookie("altcoin_session") - return resp - - -@app.post("/api/subscriptions/free-trial") -async def api_subscription_free_trial(altcoin_session: str = Cookie(default="")): - user = _require_user(altcoin_session) - try: - sub = auth_db.claim_free_trial(user["id"]) - return {"ok": True, "subscription": sub, "message": "已开通新用户免费体验1个月"} - except auth_db.AuthError as exc: - _auth_error(exc) - - -@app.get("/api/subscription/plans") -async def api_subscription_plans(): - """订阅套餐列表：当前只开放免费体验，USDT 月/季/年为预留态。""" - auth_db.init_auth_db() - conn = auth_db.get_conn() - rows = conn.execute("SELECT * FROM subscription_plan ORDER BY sort_order ASC").fetchall() - conn.close() - return [dict(r) for r in rows] - - -@app.get("/api/referral/stats") -async def api_referral_stats(altcoin_session: str = Cookie(default="")): - """当前用户的推荐统计""" - user = auth_db.get_user_by_session_token(altcoin_session) - if not user: - raise HTTPException(status_code=401, detail="请先登录") - return auth_db.get_referral_stats(user["id"]) - - -@app.get("/api/stats") -async def api_stats(altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - return get_stats() - - -@app.get("/api/recommendations") -async def api_recommendations( - limit: int = 50, - offset: int = 0, - decision_only: bool = False, - version: str = "", - paged: bool = False, - compact: bool = False, - altcoin_session: str = Cookie(default=""), -): - _require_api_user_with_subscription(altcoin_session) - # 兼容旧前端：默认仍返回数组；paged/compact 任一开启时返回分页对象。 - return get_all_recommendations( - limit, - decision_only=decision_only, - version=version, - offset=offset, - with_meta=(paged or compact), - ) - - -@app.get("/api/recommendations/active") -async def api_recommendations_active( - dedup: bool = True, - actionable_only: bool = True, - version: str = "", - hours: float = 0, - limit: int = 0, - offset: int = 0, - paged: bool = False, - compact: bool = False, - altcoin_session: str = Cookie(default=""), -): - _require_api_user_with_subscription(altcoin_session) - if dedup: - return get_active_recommendations_deduped( - actionable_only=actionable_only, - version=version, - hours=hours, - limit=limit, - offset=offset, - with_meta=(paged or compact), - ) - return get_active_recommendations(actionable_only=actionable_only) - -@app.get("/api/versions") -async def api_versions(view: str = "active", altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - """返回策略版本及当前视图数量。 - - view=active：只统计实时推荐入场计划（buy_now/wait_pullback） - view=watch：只统计观察池（observe） - 其他：统计全量 active 去重样本 - - v1.7.8 修复: 不再调用 get_active_recommendations_deduped()（该函数默认只取最新版本）。 - 改用直接 SQL 查询全量 active 去重样本，确保所有版本都出现在列表中。 - """ - # 直接 SQL: 按 (symbol, strategy_version) 去重取最新 active - # v1.7.8 fix: 同币种在不同版本都有活跃时，各版本独立计数（不在全局合并） - conn = get_conn() - rows = conn.execute(""" - SELECT r.* FROM recommendation r - JOIN ( - SELECT symbol, strategy_version, MAX(id) AS max_id - FROM recommendation - WHERE status='active' AND strategy_version IS NOT NULL AND strategy_version != '' - GROUP BY symbol, strategy_version - ) latest ON latest.max_id = r.id - ORDER BY r.strategy_version DESC, r.rec_time DESC - """).fetchall() - conn.close() - counts = {} - for row in rows: - item = dict(row) # sqlite3.Row → dict - _derive_execution_fields(item) # 派生 execution_status (原来在 get_active_recommendations_deduped 内部) - version = str(item.get("strategy_version") or "").strip() - if not version: - continue - status = item.get("execution_status") or "observe" - if view == "active" and status not in ("buy_now", "wait_pullback"): - continue - if view == "watch" and status != "observe": - continue - counts[version] = counts.get(version, 0) + 1 - - versions = [{"version": version, "count": count} for version, count in counts.items()] - - # 确保当前 rules.yaml 的版本始终在列表中，哪怕当前视图数量为0 - current_version = str(get_meta().get("strategy_version") or "").strip() - if current_version and current_version not in counts: - versions.append({"version": current_version, "count": 0}) - - def _version_key(v): - try: - parts = v["version"].lstrip("v").split(".") - return tuple(int(p) for p in parts) - except Exception: - return (0,) - versions.sort(key=_version_key, reverse=True) - return versions - - -@app.get("/api/strategy/insights") -async def api_strategy_insights(altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - return get_strategy_insights() - - -@app.get("/api/strategy/lifecycle") -async def api_strategy_lifecycle(days: int = 30, altcoin_session: str = Cookie(default="")): - """策略发布生命周期总览：研究 → 候选 → 灰度 → 正式发布。""" - _require_api_user_with_subscription(altcoin_session) - return get_strategy_iteration_dashboard(days=days) - - -@app.get("/api/personalization") -async def api_personalization(altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - return { - "watchlist": auth_db.get_watchlist_symbols(user["id"]), - "observations": auth_db.get_saved_observations(user["id"]), - "push_rules": auth_db.get_push_rules(user["id"]), - } - - -@app.post("/api/watchlist") -async def api_add_watchlist(req: WatchlistRequest, altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - auth_db.add_watchlist_symbol(user["id"], req.symbol) - return {"ok": True, "watchlist": auth_db.get_watchlist_symbols(user["id"])} - - -@app.delete("/api/watchlist/{symbol}") -async def api_remove_watchlist(symbol: str, altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - auth_db.remove_watchlist_symbol(user["id"], symbol) - return {"ok": True, "watchlist": auth_db.get_watchlist_symbols(user["id"])} - - -@app.post("/api/observations") -async def api_save_observation(req: ObservationRequest, altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - auth_db.save_observation(user["id"], req.rec_id, req.note) - return {"ok": True, "observations": auth_db.get_saved_observations(user["id"])} - - -@app.delete("/api/observations/{rec_id}") -async def api_remove_observation(rec_id: int, altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - auth_db.remove_observation(user["id"], rec_id) - return {"ok": True, "observations": auth_db.get_saved_observations(user["id"])} - - -@app.post("/api/push-rules") -async def api_update_push_rules(req: PushRulesRequest, altcoin_session: str = Cookie(default="")): - user = _require_api_user_with_subscription(altcoin_session) - rules = auth_db.update_push_rules(user["id"], req.dict()) - return {"ok": True, "push_rules": rules} - - -@app.get("/api/screening") -async def api_screening(hours: int = 24, limit: int = 100, altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - return get_screening_history(hours, limit) - - -@app.get("/api/review") -async def api_review(altcoin_session: str = Cookie(default="")): - """复盘数据 + 信号权重""" - _require_api_user_with_subscription(altcoin_session) - return get_review_stats() - - -@app.get("/api/weights") -async def api_weights(altcoin_session: str = Cookie(default="")): - """当前动态权重""" - _require_api_user_with_subscription(altcoin_session) - return get_signal_weights() - - -@app.get("/api/cron") -async def api_cron(limit: int = 50, job_name: str = "", altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - return get_cron_run_logs(limit=limit, job_name=job_name or None) - - -@app.get("/api/cron/summary") -async def api_cron_summary(hours: int = 24, altcoin_session: str = Cookie(default="")): - _require_api_user_with_subscription(altcoin_session) - return get_cron_run_summary(hours=hours) - - -@app.get("/api/sentiment") -async def api_sentiment(hours: int = 6, altcoin_session: str = Cookie(default="")): - """返回舆情监控数据：以消息/事件为主，币种作为关联信息。""" - _require_api_user_with_subscription(altcoin_session) - db = os.getenv("ALPHAX_DB_PATH", str(REPO_ROOT / "data" / "altcoin_monitor.db")) - conn = sqlite3.connect(db) - conn.row_factory = sqlite3.Row - - active_recs = conn.execute( - "SELECT DISTINCT symbol FROM recommendation WHERE status='active'" - ).fetchall() - active_symbols = {r["symbol"].split("/")[0].upper() for r in active_recs} - - recent_screened = conn.execute(""" - SELECT DISTINCT symbol FROM screening_log - WHERE scan_time >= datetime('now', '-' || ? || ' hours') - """, (hours,)).fetchall() - screened_bases = {r["symbol"].split("/")[0].upper() for r in recent_screened} - - events = [] - now_utc = datetime.now(timezone.utc) - - def _parse_event_time(value): - """解析 RSS/ISO 时间；失败返回 None，避免旧闻/脏时间混进短线舆情。""" - if not value: - return None - text = str(value).strip() - for fmt in ("%a, %d %b %Y %H:%M:%S %Z", "%a, %d %b %Y %H:%M:%S GMT"): - try: - return datetime.strptime(text, fmt).replace(tzinfo=timezone.utc) - except Exception: - pass - try: - dt = datetime.fromisoformat(text.replace("Z", "+00:00")) - if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) - return dt.astimezone(timezone.utc) - except Exception: - return None - - def _is_fresh_news(value, max_hours): - dt = _parse_event_time(value) - if not dt: - return False - age_hours = (now_utc - dt).total_seconds() / 3600 - return 0 <= age_hours <= max_hours - - valuable_news_keywords = [ - "listing", "listed", "launch", "launchpool", "megadrop", "airdrop", - "mainnet", "upgrade", "partnership", "integrat", "acquisition", "merge", - "buyback", "burn", "token burn", "funding", "raises", "investment", - "sec", "etf", "approval", "lawsuit", "settlement", "hack", "exploit", - "delist", "suspend", "roadmap", "migration", - "上线", "上币", "合约", "空投", "主网", "升级", "合作", "收购", - "回购", "销毁", "融资", "获投", "监管", "批准", "黑客", "漏洞", "下架", "暂停", - ] - low_value_news_keywords = [ - "price prediction", "price today", "live price", "marketcap and chart", - "what is", "how to buy", "good investment", "forecast", "prediction 2026", - "prediction 2027", "prediction 2030", "technical analysis", "is it time to buy", - "价格预测", "今日价格", "实时价格", "怎么买", "是什么币", - ] - - def _is_valuable_news_title(title): - text = (title or "").lower() - if not text: - return False - if any(k in text for k in low_value_news_keywords): - return False - return any(k in text for k in valuable_news_keywords) - - # 1) v1.7.3 事件驱动舆情：交易所公告 / 高时效消息 / Trending触发，天然是“消息优先”。 - try: - event_rows = conn.execute(""" - SELECT source, symbol, title, url, published_at, detected_at, importance, - event_type, decision, tech_score, rec_id, pushed - FROM event_news - WHERE detected_at >= datetime('now', '-' || ? || ' hours') - ORDER BY datetime(published_at) DESC, id DESC - LIMIT 80 - """, (hours,)).fetchall() - for r in event_rows: - base = (r["symbol"] or "").split("/")[0].upper() - source = r["source"] or "event" - title = r["title"] or "" - event_type = r["event_type"] or "event" - importance = r["importance"] or "B" - decision = r["decision"] or "" - # 纯 CoinGecko Trending 只说明“热度变化”，不是高价值舆情；舆情页不展示。 - # Trending 仍可在策略内部作为热度源参与技术确认，但不占用“有价值舆情”列表。 - if event_type == "market_heat": - continue - events.append({ - "source": source, - "source_label": "Binance公告" if "binance" in source else "CoinGecko热度" if "coingecko" in source else source, - "event_type": event_type, - "importance": importance, - "title": title, - "url": r["url"] or "", - "published_at": r["published_at"], - "detected_at": r["detected_at"], - "related_symbol": r["symbol"], - "related_base": base, - "related_name": "", - "decision": r["decision"] or "", - "tech_score": r["tech_score"] or 0, - "rec_id": r["rec_id"] or 0, - "pushed": bool(r["pushed"]), - "in_active": base in active_symbols, - "in_screened": base in screened_bases, - "price_usd": 0, - "change_24h_pct": 0, - "market_cap_rank": 0, - "trend_rank": None, - }) - except Exception: - pass - - # 2) 旧 sentiment_events：把 CoinGecko Trending 也转成“舆情消息”，不再以币种排名卡为主。 - rows = conn.execute(""" - SELECT symbol, name, trend_rank, trend_score, market_cap_rank, - detected_at, extra_json - FROM sentiment_events - WHERE detected_at = (SELECT MAX(detected_at) FROM sentiment_events WHERE source='coingecko') - ORDER BY trend_rank - """).fetchall() - for r in rows: - raw_extra = r["extra_json"] - if not raw_extra or not isinstance(raw_extra, str) or not raw_extra.strip(): - extra = {} - else: - try: - extra = json.loads(raw_extra) - except Exception: - extra = {} - base = (r["symbol"] or "").upper() - name = r["name"] or base - trend_rank = r["trend_rank"] or 0 - price_usd = extra.get("price_usd", 0) or 0 - change_24h_pct = extra.get("change_24h_pct", 0) or 0 - news_items = extra.get("news", []) or [] - - if news_items: - for n in news_items[:3]: - published = n.get("published") or "" - # 旧闻不能出现在短线舆情页：必须按新闻发布时间过滤，而不是按热度采集时间过滤。 - if not _is_fresh_news(published, hours): - continue - title = n.get("title") or f"{name} 相关新闻" - if not _is_valuable_news_title(title): - continue - events.append({ - "source": n.get("source") or "news", - "source_label": n.get("source") or "新闻", - "event_type": "news", - "importance": "B", - "title": title, - "url": n.get("url") or "", - "published_at": published, - "detected_at": r["detected_at"], - "related_symbol": f"{base}/USDT", - "related_base": base, - "related_name": name, - "decision": "", - "tech_score": 0, - "rec_id": 0, - "pushed": False, - "in_active": base in active_symbols, - "in_screened": base in screened_bases, - "price_usd": price_usd, - "change_24h_pct": change_24h_pct, - "market_cap_rank": r["market_cap_rank"], - "trend_rank": trend_rank, - }) - # 没有高价值相关新闻时，不再把“进入 Trending”本身当舆情消息展示。 - # Trending 可继续在策略内部作为热度信号参与技术确认，但舆情页只保留可解释的事件/新闻。 - - conn.close() - - # 去重：同一标题+币种只保留最新一条 - deduped = [] - seen = set() - for e in events: - key = ((e.get("title") or "").strip().lower(), e.get("related_base"), e.get("source")) - if key in seen: - continue - seen.add(key) - if e.get("in_active"): - e["relation_tag"] = "持仓/活跃推荐" - elif e.get("in_screened"): - e["relation_tag"] = "系统筛选中" - else: - e["relation_tag"] = "关联币种" - deduped.append(e) - - def _sort_key(item): - ts = item.get("published_at") or item.get("detected_at") or "" - imp = {"RISK": 5, "S": 4, "A": 3, "B": 2, "C": 1}.get(item.get("importance"), 0) - return (ts, imp) - deduped.sort(key=_sort_key, reverse=True) - - check_time = deduped[0]["detected_at"] if deduped else None - return { - "check_time": check_time, - "total_events": len(deduped), - "overlap_active": sum(1 for e in deduped if e["in_active"]), - "overlap_screened": sum(1 for e in deduped if e["in_screened"]), - "events": deduped[:80], - # 兼容旧前端/调试字段 - "trending": [], - "total_trending": 0, - } - - -@app.get("/api/kline") -async def api_kline(symbol: str, interval: str = "1d", limit: int = 60, altcoin_session: str = Cookie(default="")): - """返回 Binance K线数据（日线默认60根，前端渲染蜡烛图用）""" - _require_api_user_with_subscription(altcoin_session) - import requests as req - try: - clean = symbol.replace("/", "") - r = req.get( - "https://api.binance.com/api/v3/klines", - params={"symbol": clean, "interval": interval, "limit": limit}, - timeout=10, - ) - if r.status_code != 200: - return JSONResponse({"error": f"Binance {r.status_code}"}, status_code=502) - data = r.json() - candles = [ - { - "time": d[0], "open": float(d[1]), "high": float(d[2]), - "low": float(d[3]), "close": float(d[4]), "volume": float(d[5]), - } - for d in data - ] - return {"symbol": symbol, "interval": interval, "candles": candles} - except Exception as e: - return JSONResponse({"error": str(e)}, status_code=500) - - -@app.get("/", response_class=HTMLResponse) -async def index(): - """落地页 — 原始 HTML,无 Jinja2""" - landing_path = str(REPO_ROOT / "static" / "index.html") - with open(landing_path, "r", encoding="utf-8") as f: - return HTMLResponse(content=f.read()) - - -@app.get("/auth", response_class=HTMLResponse) -async def auth_page(): - """登录/注册页 — 原始 HTML,无 Jinja2""" - auth_path = str(REPO_ROOT / "static" / "auth.html") - with open(auth_path, "r", encoding="utf-8") as f: - return HTMLResponse(content=f.read()) - - -@app.get("/watchlist", response_class=HTMLResponse) -async def watchlist_page(request: Request): - user, redirect = _require_page_user(request) - if redirect: - return redirect - return render_page("watchlist.html", request) - - -@app.get("/strategy", response_class=HTMLResponse) -async def strategy_page(request: Request): - user, redirect = _require_page_user(request) - if redirect: - return redirect - return render_page("strategy.html", request) - - -@app.get("/subscription", response_class=HTMLResponse) -async def subscription_page(request: Request): - user, redirect = _require_page_user(request, require_subscription=False) - if redirect: - return redirect - return render_page("subscription.html", request) - - -@app.get("/referral", response_class=HTMLResponse) -async def referral_page(request: Request, altcoin_session: str = Cookie(default="")): - user, redirect = _require_page_user(request) - if redirect: - return redirect - return render_page("referral.html", request) - - -@app.get("/app", response_class=HTMLResponse) -async def app_page(altcoin_session: str = Cookie(default=""), request: Request = None): - user, redirect = _require_page_user(request) - if redirect: - return redirect - try: - auth_db.log_user_activity(user["id"], "page_view", "app", - ip=request.client.host if request and request.client else "") - except Exception: - pass - resp = templates.TemplateResponse(request=request, name="app.html", context={"show_nav": True}) - resp.headers["Cache-Control"] = "no-cache, no-store, must-revalidate" - resp.headers["Pragma"] = "no-cache" - resp.headers["Expires"] = "0" - return resp - - -def render_page(template_name: str, request: Request, **kwargs): - """统一的模板渲染,自动注入活动日志""" - try: - user = auth_db.get_user_by_session_token( - request.cookies.get("altcoin_session", "") - ) - if user: - auth_db.log_user_activity(user["id"], "page_view", template_name.replace(".html", ""), - ip=request.client.host if request.client else "") - except Exception: - pass - return templates.TemplateResponse(request=request, name=template_name, - context={"show_nav": True, **kwargs}) - - -@app.get("/api/newsfeed") -async def api_newsfeed(altcoin_session: str = Cookie(default="")): - """聚合外部新闻源：Fear & Greed Index + Google News RSS(EN+CN) + CoinGecko Trending""" - _require_api_user_with_subscription(altcoin_session) - import requests as req - import xml.etree.ElementTree as ET - from email.utils import parsedate_to_datetime - from datetime import datetime, timezone - - result = {"fear_greed": None, "trending": [], "news": []} - now = datetime.now(timezone.utc) - - # 1) Fear & Greed Index - try: - r = req.get("https://api.alternative.me/fng/?limit=1", timeout=8) - if r.status_code == 200: - fg = r.json() - d0 = fg.get("data", [{}])[0] - result["fear_greed"] = { - "value": int(d0.get("value", 50)), - "classification": d0.get("value_classification", ""), - } - except Exception: - pass - - # 2) CoinGecko trending - try: - r = req.get("https://api.coingecko.com/api/v3/search/trending", timeout=10) - if r.status_code == 200: - coins = r.json().get("coins", [])[:7] - for c in coins: - item = c.get("item", {}) - result["trending"].append({ - "name": item.get("name", ""), - "symbol": item.get("symbol", ""), - "market_cap_rank": item.get("market_cap_rank"), - "thumb": item.get("thumb", ""), - }) - except Exception: - pass - - # 3) Google News RSS - def fetch_google_news(query, hl, gl, ceid, label): - items = [] - try: - url = f"https://news.google.com/rss/search?q={req.utils.quote(query)}&hl={hl}&gl={gl}&ceid={ceid}" - r = req.get(url, timeout=12, headers={"User-Agent": "Mozilla/5.0"}) - if r.status_code != 200: - return items - root = ET.fromstring(r.text) - for el in root.findall(".//item")[:15]: - pub_str = el.findtext("pubDate", "") - dt = parsedate_to_datetime(pub_str) if pub_str else None - age_h = round((now - dt).total_seconds() / 3600, 1) if dt else None - if age_h is not None and age_h > 48: - continue - items.append({ - "title": (el.findtext("title", "") or "")[:120], - "url": el.findtext("link", "") or "", - "source": (el.findtext("source", "") or "")[:30], - "age_hours": age_h, - "lang": label, - }) - except Exception: - pass - return items - - en_news = fetch_google_news( - "cryptocurrency OR bitcoin OR ethereum OR defi OR altcoin when:24h", - "en-US", "US", "US:en", "en" - ) - cn_news = fetch_google_news( - "加密货币 OR 比特币 OR 以太坊 OR DeFi OR Web3 when:24h", - "zh-CN", "CN", "CN:zh-Hans", "cn" - ) - result["news"] = sorted(en_news + cn_news, key=lambda x: x.get("age_hours") or 999)[:30] - - return result - - -@app.get("/sentiment", response_class=HTMLResponse) -async def sentiment_page(request: Request): - user, redirect = _require_page_user(request) - if redirect: - return redirect - return render_page("sentiment.html", request) -async def stock_report_page(): - return HTMLResponse(content=STOCK_REPORT_TEMPLATE) - - -@app.get("/iteration", response_class=HTMLResponse) -async def iteration_page(request: Request): - user, redirect = _require_page_user(request) - if redirect: - return redirect - return render_page("iteration.html", request) - - -@app.get("/api/iterations") -async def api_iterations(limit: int = 30, altcoin_session: str = Cookie(default="")): - """返回策略迭代历史""" - _require_api_user_with_subscription(altcoin_session) - conn = get_conn() - rows = conn.execute( - "SELECT * FROM strategy_iteration_log ORDER BY id DESC LIMIT ?", - (limit,) - ).fetchall() - conn.close() - return [dict(r) for r in rows] - - -@app.get("/api/strategy/candidates") -async def api_strategy_candidates(limit: int = 50, status: str = "", altcoin_session: str = Cookie(default="")): - """候选/灰度规则池：研究结论先入池，达到样本门槛后再发布。""" - _require_api_user_with_subscription(altcoin_session) - return get_strategy_rule_candidates(limit=limit, status=status or None) - - -@app.get("/api/strategy/failures") -async def api_strategy_failures(limit: int = 50, altcoin_session: str = Cookie(default="")): - """失败模式归因明细。""" - _require_api_user_with_subscription(altcoin_session) - return get_strategy_failure_patterns(limit=limit) - - -@app.post("/api/strategy/candidates/refresh") -async def api_strategy_candidates_refresh(altcoin_session: str = Cookie(default="")): - """手动刷新候选规则表现与生命周期状态。""" - _require_api_user_with_subscription(altcoin_session) - return {"updated": refresh_strategy_candidate_performance()} - - -@app.get("/api/strategy/candidates/dry-run") -async def api_strategy_candidates_dry_run(altcoin_session: str = Cookie(default="")): - """候选规则表现 dry-run：只读评估，不写库、不升版。""" - _require_api_user_with_subscription(altcoin_session) - return dry_run_strategy_candidate_performance() - - -@app.post("/api/strategy/failures/backfill") -async def api_strategy_failures_backfill(dry_run: bool = False): - """历史失败模式回填：默认写库，dry_run=true时只预览。""" - return backfill_strategy_failure_patterns(dry_run=dry_run) - - -@app.post("/api/strategy/candidates/generate-history") -async def api_strategy_candidates_generate_history(dry_run: bool = False): - """从历史review_log自动生成候选规则池：默认写库，dry_run=true时只预览。""" - result = generate_candidates_from_review_history(dry_run=dry_run) - if not dry_run: - result["refreshed"] = refresh_strategy_candidate_performance() - return result - - -STOCK_REPORT_TEMPLATE = open( - REPO_ROOT / "templates" / "stock_report_template.html", - "r", - encoding="utf-8", -).read() - -HTML_PAGE = r''' - - - - -{e.detail}
返回看板", status_code=e.status_code) - return templates.TemplateResponse(request=request, name="admin.html", context={"show_nav": True}) - - -# ====== ADMIN API (v1.7.8) ====== - -@app.get("/api/admin/check") -async def api_admin_check(altcoin_session: str = Cookie(default="")): - """前端检查当前用户是否为管理员""" - try: - user = _require_admin(altcoin_session) - return {"is_admin": True, "email": user.get("email", "")} - except HTTPException: - return {"is_admin": False} - - -@app.get("/api/admin/stats") -async def api_admin_stats(altcoin_session: str = Cookie(default="")): - _require_admin(altcoin_session) - return auth_db.get_admin_stats() - - -@app.get("/api/admin/users") -async def api_admin_users(search: str = "", offset: int = 0, limit: int = 50, tab: str = "all", - altcoin_session: str = Cookie(default="")): - _require_admin(altcoin_session) - return auth_db.get_admin_users(search=search, offset=offset, limit=limit, tab=tab) - - -@app.get("/api/admin/orders") -async def api_admin_orders(search: str = "", offset: int = 0, limit: int = 50, status: str = "all", - altcoin_session: str = Cookie(default="")): - _require_admin(altcoin_session) - return auth_db.get_admin_orders(search=search, offset=offset, limit=limit, status=status) + current_request.reset(token) diff --git a/docker/scheduler.py b/docker/scheduler.py index 4eab4e3..481ccbe 100755 --- a/docker/scheduler.py +++ b/docker/scheduler.py @@ -25,7 +25,7 @@ DRY_RUN = os.getenv("ALPHAX_SCHEDULER_DRY_RUN", "1").strip() not in {"0", "false @dataclass class Job: name: str - module: str + command: str every_seconds: int args: tuple[str, ...] = () initial_delay: int = 0 @@ -43,7 +43,7 @@ def env_for_child() -> dict[str, str]: def run_job(job: Job) -> None: - cmd = [PYTHON, "-m", job.module, *job.args] + cmd = [PYTHON, "-m", "app.cli", job.command, *job.args] print(f"[{now_str()}] [scheduler] start {job.name}: {' '.join(cmd)}", flush=True) if DRY_RUN: print(f"[{now_str()}] [scheduler] DRY_RUN=1 skip {job.name}", flush=True) @@ -75,12 +75,12 @@ def run_job(job: Job) -> None: def build_jobs() -> list[Job]: # 与当前宿主机 crontab 对齐，但串行执行。 return [ - Job("event", "app.services.event_driven_screener", 60, ("--once",), initial_delay=5), - Job("tracker", "app.services.price_tracker", 180, initial_delay=20), - Job("confirm", "app.services.altcoin_confirm", 600, initial_delay=40), - Job("screener", "app.services.altcoin_screener", 900, initial_delay=80), - Job("sentiment", "app.services.sentiment_monitor", 1800, ("--collect",), initial_delay=120), - Job("review", "app.services.review_engine", 24 * 3600, initial_delay=300), + Job("event", "event", 60, initial_delay=5), + Job("tracker", "tracker", 180, initial_delay=20), + Job("confirm", "confirm", 600, initial_delay=40), + Job("screener", "screener", 900, initial_delay=80), + Job("sentiment", "sentiment", 1800, ("--collect",), initial_delay=120), + Job("review", "review", 24 * 3600, initial_delay=300), ] diff --git a/docs/PROJECT_STRUCTURE.md b/docs/PROJECT_STRUCTURE.md index d28560b..d93858a 100644 --- a/docs/PROJECT_STRUCTURE.md +++ b/docs/PROJECT_STRUCTURE.md @@ -45,6 +45,28 @@ - `app/integrations/feishu_push.py` - `app/integrations/feishu_review_push.py` +当前已新增的结构化外观层： + +- `app/web/routes_auth.py` +- `app/web/routes_recommendations.py` +- `app/web/routes_strategy.py` +- `app/web/routes_admin.py` +- `app/web/routes_pages.py` +- `app/web/routes_content.py` +- `app/web/shared.py` + +当前已新增的 DB 分组接口层： + +- `app/db/schema.py` +- `app/db/recommendation_queries.py` +- `app/db/review_queries.py` +- `app/db/analytics.py` +- `app/db/admin_queries.py` + +当前已新增的统一命令入口： + +- `app/cli.py` + ### 4. 配置与部署 - `rules.yaml` @@ -106,10 +128,9 @@ 如果继续整理，建议下一步做实现层拆分，而不是再搬目录： -1. 拆 `web_server.py` -2. 拆 `altcoin_db.py` -3. 为 `rules.yaml` 增加 schema 校验 -4. 为主脚本建立统一 CLI 入口 +1. 继续拆 `altcoin_db.py` +2. 深化 `rules.yaml` 的子字段 schema 校验 +3. 让 `docker/` 和文档统一收敛到 `app/cli.py` ## 当前结论 diff --git a/legacy/price_tracker_ws.py b/legacy/price_tracker_ws.py index bae0a39..b785d7a 100644 --- a/legacy/price_tracker_ws.py +++ b/legacy/price_tracker_ws.py @@ -22,6 +22,7 @@ from app.db.altcoin_db import ( log_push, ) from app.integrations.feishu_push import push_altcoin_tp_sl_alert +from app.core.opportunity_lifecycle import apply_entry_quality_gate POLL_INTERVAL = 5 REFRESH_INTERVAL = 60 @@ -49,7 +50,7 @@ def check_triggers(symbol, rec, current_price): action, signals = detect_candidate_action(rec, current_price, {}) if not action: return None - return {"action_status": action, "signals": signals} + return {"action_status": action, "signals": signals, "pushable": action == "可即刻买入"} def detect_candidate_action(rec, current_price, track_result): @@ -68,14 +69,40 @@ def detect_candidate_action(rec, current_price, track_result): entry_price = rec.get("entry_price", 0) or 0 cur_action = rec.get("action_status", "持有") + candidate_action = None + candidate_signals = [] if entry_action == "等回踩" and plan_entry_price > 0 and current_price <= plan_entry_price: ep["entry_trigger_confirmed"] = True - return "可即刻买入", [f"回踩到位 ${plan_entry_price:.6f}"] - if entry_action in ("即刻买入", "可即刻买入") and current_price <= (plan_entry_price or entry_price): + candidate_action = "可即刻买入" + candidate_signals = [f"回踩到位 ${plan_entry_price:.6f}"] + elif entry_action in ("即刻买入", "可即刻买入") and current_price <= (plan_entry_price or entry_price): ep["entry_trigger_confirmed"] = True - return "可即刻买入", ["入场条件仍满足"] - if cur_action == "可即刻买入": - return "可即刻买入", ["入场窗口延续"] + candidate_action = "可即刻买入" + candidate_signals = ["入场条件仍满足"] + elif cur_action == "可即刻买入": + candidate_action = "可即刻买入" + candidate_signals = ["入场窗口延续"] + + if candidate_action: + raw_signals = rec.get("signals", []) + if isinstance(raw_signals, str): + try: + raw_signals = json.loads(raw_signals) + except Exception: + raw_signals = [raw_signals] if raw_signals else [] + gated_action, gated_plan, reasons = apply_entry_quality_gate( + action_status=candidate_action, + entry_plan=ep, + signals=raw_signals, + current_price=current_price, + market_context=rec.get("market_context") or {}, + derivatives_context=rec.get("derivatives_context") or {}, + sector_context=rec.get("sector_context") or {}, + ) + if gated_action != "可即刻买入": + reason = reasons[0] if reasons else "买点质量闸门降级" + return gated_action, [reason] + return gated_action, candidate_signals return None, [] diff --git a/requirements.txt b/requirements.txt index d07918f..db6569e 100644 --- a/requirements.txt +++ b/requirements.txt @@ -8,3 +8,5 @@ PyYAML==6.0.2 pydantic==2.10.4 python-multipart==0.0.20 pytest==8.3.4 +httpx==0.28.1 +jinja2==3.1.6 diff --git a/static/app.html b/static/app.html index 7b30e6e..7b4a708 100644 --- a/static/app.html +++ b/static/app.html @@ -1,5 +1,5 @@ {% extends "base.html" %} -{% block title %}AlphaX — 看板{% endblock %} +{% block title %}AlphaX / Omnix — 看板{% endblock %} {% block extra_head_css %} @@ -248,9 +248,10 @@ {% block content %}