From 23a9a5175ed4abecfdcb89d5123b227580120be8 Mon Sep 17 00:00:00 2001 From: eezd Date: Fri, 16 Oct 2026 14:47:06 +0800 Subject: [PATCH] Record news signals forward for a no-trade edge test The trader labels crypto news published after 2026-10-16 15:00 with the configured model (category and BTC direction) and, four hours later, backfills BTC/ETH 1m opens at publish, +5m, +1h and +4h from the klines table. A drawer on the trading page shows the cumulative returns of trading the labelled direction, after fees. Nothing is traded; this only collects an out-of-sample record. --- .../20261016_c5e8a1b3d207_news_signals.py | 39 +++++ backend/src/alphapulse/api/trading.py | 61 +++++++- backend/src/alphapulse/models/__init__.py | 3 +- backend/src/alphapulse/models/trading.py | 16 ++ backend/src/alphapulse/trading/engine.py | 4 +- .../src/alphapulse/trading/news_signals.py | 144 ++++++++++++++++++ .../src/alphapulse/trading/signal_stats.py | 54 +++++++ backend/tests/test_signal_stats.py | 32 ++++ web/src/apis/trading.ts | 39 +++++ .../trading/components/NewsSignalDrawer.vue | 136 +++++++++++++++++ web/src/pages/trading/index.vue | 16 +- 11 files changed, 537 insertions(+), 7 deletions(-) create mode 100644 backend/alembic/versions/20261016_c5e8a1b3d207_news_signals.py create mode 100644 backend/src/alphapulse/trading/news_signals.py create mode 100644 backend/src/alphapulse/trading/signal_stats.py create mode 100644 backend/tests/test_signal_stats.py create mode 100644 web/src/pages/trading/components/NewsSignalDrawer.vue diff --git a/backend/alembic/versions/20261016_c5e8a1b3d207_news_signals.py b/backend/alembic/versions/20261016_c5e8a1b3d207_news_signals.py new file mode 100644 index 0000000..ddf8208 --- /dev/null +++ b/backend/alembic/versions/20261016_c5e8a1b3d207_news_signals.py @@ -0,0 +1,39 @@ +"""news signals + +Revision ID: c5e8a1b3d207 +Revises: a7d3e5f20c18 +Create Date: 2026-10-16 16:00:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa +from sqlalchemy.dialects import postgresql + + +# revision identifiers, used by Alembic. +revision: str = 'c5e8a1b3d207' +down_revision: Union[str, Sequence[str], None] = 'a7d3e5f20c18' +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + """Upgrade schema.""" + op.create_table( + 'news_signals', + sa.Column('news_id', sa.BigInteger(), nullable=False), + sa.Column('category', sa.String(length=16), nullable=False), + sa.Column('direction', sa.String(length=8), nullable=False), + sa.Column('model', sa.String(length=64), nullable=False), + sa.Column('labeled_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False), + sa.Column('prices', postgresql.JSONB(astext_type=sa.Text()), nullable=True), + sa.ForeignKeyConstraint(['news_id'], ['news_items.id']), + sa.PrimaryKeyConstraint('news_id'), + ) + + +def downgrade() -> None: + """Downgrade schema.""" + op.drop_table('news_signals') diff --git a/backend/src/alphapulse/api/trading.py b/backend/src/alphapulse/api/trading.py index e6ef709..f5601fb 100644 --- a/backend/src/alphapulse/api/trading.py +++ b/backend/src/alphapulse/api/trading.py @@ -17,11 +17,13 @@ from alphapulse.api.market import call_binance, client as market_client from alphapulse.core.auth import require_admin from alphapulse.core.db import get_session from alphapulse.core.redis import KEY_PREFIX, redis -from alphapulse.models import AiProvider, TradingAccount, TradingAnalysis, TradingEvent, TradingFill, TradingOrder, TradingSymbol +from alphapulse.models import AiProvider, NewsItem, NewsSignal, TradingAccount, TradingAnalysis, TradingEvent, TradingFill, TradingOrder, TradingSymbol from alphapulse.trading.client import OWN_CLIENT_ID, ExchangeError, public_get from alphapulse.trading.executor import MANUAL_ACTION_KEY +from alphapulse.trading.news_signals import FORWARD_START, SYMBOLS as SIGNAL_SYMBOLS, news_text from alphapulse.trading.positions import build_positions from alphapulse.trading.plan import DAILY_ANALYSIS_LIMIT +from alphapulse.trading.signal_stats import Signal, summarize from alphapulse.trading.snapshot import read_snapshot router = APIRouter(prefix="/api/trading", tags=["trading"], dependencies=[Depends(require_admin)]) @@ -179,6 +181,37 @@ class PositionHistoryPage(BaseModel): items: list[PositionHistoryItem] +class NewsSignalStat(BaseModel): + category: str + symbol: str + horizon: str + n: int + mean_bp: float | None + t: float | None + # 扣除来回手续费后的平均收益 + net_bp: float | None + win_rate: float | None + + +class NewsSignalItem(BaseModel): + news_id: int + published_at: datetime + source_code: str + title: str + category: str + direction: str + # 进场后的原始涨跌（bp，未按方向取正负）：{symbol: {"1h": ..., "4h": ...}}；未满 4 小时为空 + returns: dict[str, dict[str, float]] | None + + +class NewsSignalsOut(BaseModel): + start: datetime + labeled: int + filled: int + stats: list[NewsSignalStat] + recent: list[NewsSignalItem] + + async def get_account(session: AsyncSession) -> TradingAccount: return await session.get(TradingAccount, 1) or TradingAccount(id=1) @@ -454,6 +487,32 @@ async def list_position_history_symbols(session: Annotated[AsyncSession, Depends return list(await session.scalars(select(TradingFill.symbol).distinct().order_by(TradingFill.symbol))) +@router.get("/news-signals") +async def list_news_signals(session: Annotated[AsyncSession, Depends(get_session)]) -> NewsSignalsOut: + """新闻信号前向记录：按标注方向交易的累计统计和最近 50 条""" + rows = (await session.execute( + select(NewsSignal, NewsItem.published_at, NewsItem.source_code, news_text()) + .join(NewsItem, NewsItem.id == NewsSignal.news_id).order_by(NewsItem.published_at) + )).all() + filled = [Signal(published_at, s.category, s.direction, s.prices) for s, published_at, _, _ in rows if s.prices] + horizons = ("1h", "4h") + + def returns(prices: dict | None) -> dict[str, dict[str, float]] | None: + if prices is None: + return None + return {sym: {h: (float(p[h]) / float(p["entry"]) - 1) * 1e4 for h in horizons} for sym, p in prices.items()} + + return NewsSignalsOut( + start=FORWARD_START, labeled=len(rows), filled=len(filled), + stats=[NewsSignalStat(**r) for r in summarize(filled, SIGNAL_SYMBOLS, horizons)], + recent=[ + NewsSignalItem(news_id=s.news_id, published_at=published_at, source_code=source, title=title, + category=s.category, direction=s.direction, returns=returns(s.prices)) + for s, published_at, source, title in reversed(rows[-50:]) + ], + ) + + @router.post("/symbols/{symbol}/analyze", status_code=status.HTTP_202_ACCEPTED) async def request_analysis(symbol: str, session: Annotated[AsyncSession, Depends(get_session)]) -> None: """排队一次手动分析：该币停止交易时为试运行，计划只记录、不下单""" diff --git a/backend/src/alphapulse/models/__init__.py b/backend/src/alphapulse/models/__init__.py index 66e3a2a..4d1b70c 100644 --- a/backend/src/alphapulse/models/__init__.py +++ b/backend/src/alphapulse/models/__init__.py @@ -2,7 +2,7 @@ from alphapulse.core.db import Base from alphapulse.models.ai import AiProvider, ChatMessage, ChatSession, ContextVersion, SkillNode, SkillVersion from alphapulse.models.market import Kline, KlineSyncState, MarketSymbol from alphapulse.models.news import CrawlLog, CrawlRun, NewsItem, NewsRevision, RawBody, RawResponse, Source -from alphapulse.models.trading import TradingAccount, TradingAnalysis, TradingEvent, TradingFill, TradingOrder, TradingSymbol +from alphapulse.models.trading import NewsSignal, TradingAccount, TradingAnalysis, TradingEvent, TradingFill, TradingOrder, TradingSymbol __all__ = [ "AiProvider", @@ -16,6 +16,7 @@ __all__ = [ "KlineSyncState", "MarketSymbol", "NewsItem", + "NewsSignal", "NewsRevision", "RawBody", "RawResponse", diff --git a/backend/src/alphapulse/models/trading.py b/backend/src/alphapulse/models/trading.py index e9da887..a0065c5 100644 --- a/backend/src/alphapulse/models/trading.py +++ b/backend/src/alphapulse/models/trading.py @@ -145,3 +145,19 @@ class TradingFill(Base): commission: Mapped[Decimal] = mapped_column(Numeric) commission_asset: Mapped[str] = mapped_column(String(16)) time: Mapped[datetime] = mapped_column(DateTime(timezone=True)) + + +class NewsSignal(Base): + """新闻信号前向记录：加密相关新闻的模型标注，以及发布后 BTC/ETH 的价格，检验按标注方向交易是否有优势。只记录，不下单""" + + __tablename__ = "news_signals" + + news_id: Mapped[int] = mapped_column(BigInteger, ForeignKey("news_items.id"), primary_key=True) + # price_report（行情播报）/ event（事件）/ irrelevant（与币价无关） + category: Mapped[str] = mapped_column(String(16)) + # 对 BTC 短期（几小时）的影响：bullish / bearish / neutral + direction: Mapped[str] = mapped_column(String(8)) + model: Mapped[str] = mapped_column(String(64)) + labeled_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + # 发布 4 小时后回填的 1m 开盘价：{symbol: {"open": 发布所在分钟, "entry": +5 分钟, "1h": +65 分钟, "4h": +245 分钟}}；为空表示尚未回填 + prices: Mapped[dict | None] = mapped_column(JSONB) diff --git a/backend/src/alphapulse/trading/engine.py b/backend/src/alphapulse/trading/engine.py index 6e56f5b..ca6341a 100644 --- a/backend/src/alphapulse/trading/engine.py +++ b/backend/src/alphapulse/trading/engine.py @@ -1,4 +1,4 @@ -"""模拟盘交易进程：同步账户快照、执行器对账、调度分析。 +"""模拟盘交易进程：同步账户快照、执行器对账、调度分析、记录新闻信号。 用法：python -m alphapulse.trading.engine """ @@ -13,6 +13,7 @@ from alphapulse.core.db import SessionLocal from alphapulse.models import TradingAnalysis, TradingSymbol from alphapulse.trading.analyst import run_analysis from alphapulse.trading.executor import executor_loop +from alphapulse.trading.news_signals import news_signal_loop from alphapulse.trading.schedule import routine_requests from alphapulse.trading.snapshot import snapshot_loop @@ -74,6 +75,7 @@ async def main() -> None: tg.create_task(snapshot_loop()) tg.create_task(executor_loop()) tg.create_task(analysis_loop()) + tg.create_task(news_signal_loop()) if __name__ == "__main__": diff --git a/backend/src/alphapulse/trading/news_signals.py b/backend/src/alphapulse/trading/news_signals.py new file mode 100644 index 0000000..c6890a4 --- /dev/null +++ b/backend/src/alphapulse/trading/news_signals.py @@ -0,0 +1,144 @@ +"""新闻信号前向记录：给新进来的加密相关新闻打标签（类别、对 BTC 的方向），发布 4 小时后回填 BTC/ETH 的价格。 + +研究（2026-10-16，见 CONTEXT.md）：事件新闻按模型标注方向持有 4 小时，样本内约 +15bp 但不显著，且模型可能记得那段行情。 +前向记录只看标注时还没发生的行情，用来确认或否定这条线索。只记录，不下单。 +""" + +import asyncio +import logging +import re +from datetime import UTC, datetime, timedelta + +from sqlalchemy import func, select +from sqlalchemy.dialects.postgresql import insert + +from alphapulse.ai import llm +from alphapulse.ai.tools import LOCAL_TZ +from alphapulse.core.db import SessionLocal +from alphapulse.models import AiProvider, Kline, NewsItem, NewsSignal, TradingAccount + +logger = logging.getLogger("alphapulse.trading.news_signals") + +# 研究样本截止到 2026-10-16 14:15，之后发布的新闻才计入，两份样本互不重叠 +FORWARD_START = datetime(2026, 10, 16, 15, 0, tzinfo=LOCAL_TZ) +CRYPTO_PATTERN = "比特币|加密货币|以太坊|加密市场|数字货币|稳定币|Coinbase|微策略|币安|BTC" +SYMBOLS = ("BTCUSDT", "ETHUSDT") +# 发布所在分钟之后第几分钟的 1m 开盘价：发布后第 5 分钟进场（抓取加分析约 1–3 分钟），持有 1h、4h +OFFSETS = {"open": 0, "entry": 5, "1h": 65, "4h": 245} +LOOP_SECONDS = 60 +BATCH = 40 +CATEGORIES = {"price_report", "event", "irrelevant"} +DIRECTIONS = {"bullish", "bearish", "neutral"} + +SYSTEM = """你是加密货币新闻分类员。只根据给出的新闻文字判断，不要使用你对之后行情的任何记忆。 +对每条新闻给出： +- category：price_report（只是在报道价格已经发生的涨跌，如「比特币跌破 78000 美元」「比特币日内涨超 4%」，包括概念股跟随币价的涨跌）； + event（会影响币价的事件或信息，如机构增持/减持、ETF 资金流、监管政策、宏观政策对加密市场的影响、黑客攻击、交易所动态）； + irrelevant（与 BTC/ETH 价格基本无关，如个别公司财报细节、分析师调整目标价、技术合作）。 +- direction：对 BTC 价格短期（几小时内）的影响，bullish / bearish / neutral。price_report 按它报道的涨跌方向填。 +用 label 工具一次提交全部结果，id 与输入一一对应。""" +TOOL = llm.ToolSpec("label", "提交分类结果", { + "type": "object", + "properties": {"items": {"type": "array", "items": { + "type": "object", + "properties": { + "id": {"type": "integer"}, + "category": {"type": "string", "enum": sorted(CATEGORIES)}, + "direction": {"type": "string", "enum": sorted(DIRECTIONS)}, + }, + "required": ["id", "category", "direction"], + }}}, + "required": ["items"], +}) + + +def news_text(): + """快讯常常没有标题，取正文开头""" + return func.coalesce(func.nullif(NewsItem.title, ""), func.left(NewsItem.content_text, 120)) + + +def clean(text: str) -> str: + # 去掉「财联社10月7日电，」之类的日期前缀，减少模型联想到具体日期 + return re.sub(r"^(财联社)?\d{1,2}月\d{1,2}日电[，,]?", "", text).strip()[:200] + + +async def label_new() -> int: + """给还没打标签的加密新闻打标签，返回本轮标注条数。模型漏掉的条目下一轮重试""" + async with SessionLocal() as session: + rows = (await session.execute( + select(NewsItem.id, news_text()).outerjoin(NewsSignal, NewsSignal.news_id == NewsItem.id) + .where(NewsSignal.news_id.is_(None), NewsItem.published_at >= FORWARD_START, news_text().op("~")(CRYPTO_PATTERN)) + .order_by(NewsItem.published_at).limit(BATCH) + )).all() + if not rows: + return 0 + account = await session.get(TradingAccount, 1) + provider = await session.get(AiProvider, account.provider_id) if account and account.provider_id else await session.scalar( + select(AiProvider).where(AiProvider.is_default) + ) + if provider is None: + raise RuntimeError("没有可用的模型配置，无法给新闻打标签") + session.expunge(provider) + + completion = None + text = "\n".join(f"{news_id}\t{clean(t)}" for news_id, t in rows) + async for item in llm.stream(provider, SYSTEM, [{"role": "user", "text": text}], [TOOL], "low"): + if isinstance(item, llm.Completion): + completion = item + call = next((c for c in completion.tool_calls if c["name"] == "label"), None) + if call is None: + raise RuntimeError("模型没有调用 label 工具") + ids = {news_id for news_id, _ in rows} + values = [ + {"news_id": x["id"], "category": x["category"], "direction": x["direction"], "model": provider.model} + for x in call["arguments"].get("items", []) + if x.get("id") in ids and x.get("category") in CATEGORIES and x.get("direction") in DIRECTIONS + ] + if len(values) < len(rows): + logger.warning("新闻标签：%d 条中模型只返回了 %d 条有效结果，其余下一轮重试", len(rows), len(values)) + if values: + async with SessionLocal() as session: + await session.execute(insert(NewsSignal).values(values).on_conflict_do_nothing()) + await session.commit() + return len(values) + + +async def fill_prices() -> int: + """发布满 4 小时的信号回填价格，返回回填条数。1m K 线有缺口时留到下一轮""" + cutoff = datetime.now(UTC) - timedelta(minutes=OFFSETS["4h"] + 2) + filled = 0 + async with SessionLocal() as session: + pending = (await session.execute( + select(NewsSignal, NewsItem.published_at).join(NewsItem, NewsItem.id == NewsSignal.news_id) + .where(NewsSignal.prices.is_(None), NewsItem.published_at <= cutoff) + )).all() + for signal, published_at in pending: + minute = int(published_at.timestamp()) // 60 * 60_000 + times = {key: minute + offset * 60_000 for key, offset in OFFSETS.items()} + prices = {} + for symbol in SYMBOLS: + opens = dict((await session.execute( + select(Kline.open_time, Kline.open).where( + Kline.market == "um", Kline.symbol == symbol, Kline.interval == "1m", Kline.open_time.in_(times.values()), + ) + )).tuples().all()) + if len(opens) < len(times): + break + prices[symbol] = {key: str(opens[t]) for key, t in times.items()} + else: + signal.prices = prices + filled += 1 + await session.commit() + return filled + + +async def news_signal_loop() -> None: + while True: + try: + labeled = await label_new() + filled = await fill_prices() + if labeled or filled: + logger.info("新闻信号：标注 %d 条，回填价格 %d 条", labeled, filled) + except Exception: + logger.exception("新闻信号记录出错，稍后重试") + await asyncio.sleep(LOOP_SECONDS) diff --git a/backend/src/alphapulse/trading/signal_stats.py b/backend/src/alphapulse/trading/signal_stats.py new file mode 100644 index 0000000..20018df --- /dev/null +++ b/backend/src/alphapulse/trading/signal_stats.py @@ -0,0 +1,54 @@ +"""新闻信号的统计（纯函数，便于测试）：按模型标注方向交易的收益，同一件事的多条报道只算一次""" + +import math +from dataclasses import dataclass +from datetime import datetime, timedelta + +# 多个来源报道同一件事：同一类别 10 分钟内只算第一条 +DEDUPE = timedelta(minutes=10) +# 吃单进出各 0.04% +FEE_BP = 8.0 +DIRECTION_SIGN = {"bullish": 1, "bearish": -1} + + +@dataclass +class Signal: + published_at: datetime + category: str + direction: str + # {symbol: {"entry": ..., "1h": ..., "4h": ...}}，价格为十进制字符串 + prices: dict + + +def dedupe(signals: list[Signal]) -> list[Signal]: + """signals 按发布时间排序""" + last: dict[str, datetime] = {} + kept = [] + for s in signals: + if s.category not in last or s.published_at - last[s.category] > DEDUPE: + kept.append(s) + last[s.category] = s.published_at + return kept + + +def summarize(signals: list[Signal], symbols: tuple[str, ...], horizons: tuple[str, ...]) -> list[dict]: + """每个类别、币、持有时长：利多做多、利空做空的收益（bp），中性不交易""" + rows = [] + kept = dedupe(signals) + for category in sorted({s.category for s in kept}): + for symbol in symbols: + for horizon in horizons: + pnl = [ + DIRECTION_SIGN[s.direction] * (float(s.prices[symbol][horizon]) / float(s.prices[symbol]["entry"]) - 1) * 1e4 + for s in kept if s.category == category and s.direction in DIRECTION_SIGN + ] + n = len(pnl) + mean = sum(pnl) / n if n else None + sd = math.sqrt(sum((x - mean) ** 2 for x in pnl) / (n - 1)) if n > 1 else None + rows.append({ + "category": category, "symbol": symbol, "horizon": horizon, "n": n, + "mean_bp": mean, "t": mean / (sd / math.sqrt(n)) if sd else None, + "net_bp": mean - FEE_BP if mean is not None else None, + "win_rate": sum(x > 0 for x in pnl) / n if n else None, + }) + return rows diff --git a/backend/tests/test_signal_stats.py b/backend/tests/test_signal_stats.py new file mode 100644 index 0000000..3c931b5 --- /dev/null +++ b/backend/tests/test_signal_stats.py @@ -0,0 +1,32 @@ +from datetime import datetime, timedelta + +import pytest + +from alphapulse.trading.signal_stats import Signal, dedupe, summarize + +T0 = datetime(2026, 10, 16, 15, 0) + + +def signal(minutes: float, category: str, direction: str, entry: str = "100", after: str = "101") -> Signal: + return Signal(T0 + timedelta(minutes=minutes), category, direction, {"BTCUSDT": {"entry": entry, "1h": after, "4h": after}}) + + +def test_dedupe_keeps_first_report_of_same_story_per_category(): + kept = dedupe([signal(0, "event", "bullish"), signal(5, "event", "bearish"), signal(5, "price_report", "bullish"), signal(11, "event", "bullish")]) + assert [(s.published_at - T0, s.category) for s in kept] == [ + (timedelta(0), "event"), (timedelta(minutes=5), "price_report"), (timedelta(minutes=11), "event"), + ] + + +def test_summarize_trades_label_direction_and_skips_neutral(): + rows = summarize([ + signal(0, "event", "bullish", after="101"), # 做多 +100bp + signal(20, "event", "bearish", after="101"), # 做空 −100bp + signal(40, "event", "bearish", after="98"), # 做空 +200bp + signal(60, "event", "neutral", after="150"), # 中性不交易 + ], ("BTCUSDT",), ("4h",)) + assert len(rows) == 1 + row = rows[0] + assert (row["n"], row["win_rate"]) == (3, pytest.approx(2 / 3)) + assert row["mean_bp"] == pytest.approx(200 / 3) + assert row["net_bp"] == pytest.approx(200 / 3 - 8) diff --git a/web/src/apis/trading.ts b/web/src/apis/trading.ts index 329ff15..2964646 100644 --- a/web/src/apis/trading.ts +++ b/web/src/apis/trading.ts @@ -319,6 +319,45 @@ export function listPositionHistorySymbols(): Promise { return http.get("/trading/positions/history/symbols") } +export type SignalHorizon = "1h" | "4h" + +/** 新闻信号前向记录：按模型标注方向（利多做多、利空做空）交易的统计；收益单位 bp */ +export interface NewsSignalStat { + category: "price_report" | "event" | "irrelevant" + symbol: string + horizon: SignalHorizon + n: number + mean_bp: number | null + t: number | null + /** 扣除来回手续费后的平均收益 */ + net_bp: number | null + win_rate: number | null +} + +export interface NewsSignalItem { + news_id: number + published_at: string + source_code: string + title: string + category: NewsSignalStat["category"] + direction: "bullish" | "bearish" | "neutral" + /** 进场后的原始涨跌（bp，未按方向取正负）；发布未满 4 小时为空 */ + returns: Record> | null +} + +export interface NewsSignals { + /** 前向记录的起点：之后发布的新闻才计入 */ + start: string + labeled: number + filled: number + stats: NewsSignalStat[] + recent: NewsSignalItem[] +} + +export function getNewsSignals(): Promise { + return http.get("/trading/news-signals") +} + export function getSymbolPlan(symbol: string): Promise<{ plan: Analysis | null, orders: OrderRow[] }> { return http.get(`/trading/symbols/${symbol}/plan`) } diff --git a/web/src/pages/trading/components/NewsSignalDrawer.vue b/web/src/pages/trading/components/NewsSignalDrawer.vue new file mode 100644 index 0000000..8c1f7da --- /dev/null +++ b/web/src/pages/trading/components/NewsSignalDrawer.vue @@ -0,0 +1,136 @@ + + + diff --git a/web/src/pages/trading/index.vue b/web/src/pages/trading/index.vue index 7127ef4..0072b9b 100644 --- a/web/src/pages/trading/index.vue +++ b/web/src/pages/trading/index.vue @@ -1,13 +1,14 @@