Align backend authentication and API response contracts

Replace cookie authentication with revocable Bearer sessions, optional one-use PNG challenges, bounded login limits, and short-lived market WebSocket tickets. Keep captcha disabled by default.

Standardize ordinary JSON responses while preserving HTTP statuses, raw OpenAPI, redirects, SSE, and WebSocket semantics. Revoke long connections with their owning session.

Expose official calendar data and collector status to the frontend and research tools, support bilingual news topics, and keep summaries and forward-only labels explicit. Verify the staged snapshot with all 305 backend tests.

backend/tests/test_auth_protocol.py

@@ -0,0 +1,360 @@
+"""认证和公开接口契约；Redis/数据库/外部服务均隔离。"""
+import asyncio
+import base64
+import io
+import time
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import httpx
+import jwt
+import pytest
+from fastapi.testclient import TestClient
+from PIL import Image
+from redis.exceptions import ConnectionError
+from starlette.testclient import WebSocketDenialResponse
+
+
+class MemoryRedis:
+    def __init__(self):
+        self.values = {}
+        self.offset = 0
+        self.failed = False
+
+    def now(self):
+        if self.failed:
+            raise ConnectionError("isolated test failure")
+        return time.time() + self.offset
+
+    async def set(self, key, value, *, ex=None, exat=None):
+        now = self.now()
+        self.values[key] = (value, exat if exat is not None else now + ex)
+
+    async def get(self, key):
+        now = self.now()
+        value = self.values.get(key)
+        if value is None:
+            return None
+        if value[1] <= now:
+            self.values.pop(key)
+            return None
+        return value[0]
+
+    async def getdel(self, key):
+        value = await self.get(key)
+        self.values.pop(key, None)
+        return value
+
+    async def delete(self, key):
+        self.now()
+        self.values.pop(key, None)
+
+    async def ttl(self, key):
+        return max(0, int(self.values[key][1] - self.now())) if await self.get(key) else -2
+
+    async def eval(self, script, numkeys, key, seconds, threshold):
+        current = int(await self.get(key) or 0) + 1
+        expires = self.now() + seconds if current in (1, threshold) else self.values[key][1]
+        self.values[key] = (str(current), expires)
+        return current
+
+
+@pytest.fixture
+def auth_app(trading, monkeypatch):
+    from alphapulse.api import auth
+    from alphapulse.api.main import app
+    from alphapulse.core import auth as core
+    settings = core.get_settings()
+    monkeypatch.setattr(settings, "jwt_secret", "isolated-auth-test-secret-at-least-32-bytes")
+    monkeypatch.setattr(settings, "admin_username", "test-admin")
+    monkeypatch.setattr(settings, "admin_password_hash", core._hasher.hash("test-password"))
+    monkeypatch.setattr(settings, "login_captcha_enabled", True)
+    memory = MemoryRedis()
+    monkeypatch.setattr(auth, "redis", memory)
+    monkeypatch.setattr(core, "redis", memory)
+    yield app, memory
+    app.dependency_overrides.clear()
+
+
+async def challenge(client):
+    result = await client.get("/api/auth/captcha")
+    assert result.status_code == 200
+    body = result.json()
+    assert body["code"] == 0 and body["message"] == ""
+    return body["data"]
+
+
+async def login(client, memory, *, password="test-password"):
+    from alphapulse.api.auth import captcha_digest
+    from alphapulse.core.auth import AUTH_PREFIX
+    data = await challenge(client)
+    await memory.set(f"{AUTH_PREFIX}captcha:{data['captcha_id']}", captcha_digest(data["captcha_id"], "ABCDE"), ex=120)
+    return await client.post("/api/auth/login", json={"username": "test-admin", "password": password,
+                                                    "captcha_id": data["captcha_id"], "code": "ABCDE"})
+
+
+def test_login_png_restore_revoke_current_session_only(auth_app):
+    app, memory = auth_app
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            image = await challenge(client)
+            with Image.open(io.BytesIO(base64.b64decode(image["image"].split(",")[1]))) as png:
+                assert png.format == "PNG" and png.size == (180, 60)
+            assert set(image) == {"enabled", "captcha_id", "image", "expires_in"}
+            assert image["enabled"] is True
+            first, second = await login(client, memory), await login(client, memory)
+            assert first.status_code == second.status_code == 200
+            token = first.json()["data"]
+            assert token["token_type"] == "bearer" and token["expires_at"]
+            assert "set-cookie" not in first.headers
+            headers = {"Authorization": f"Bearer {token['token']}"}
+            assert (await client.get("/api/auth/me", headers=headers)).json()["data"] == {"username": "test-admin", "is_admin": True}
+            assert (await client.post("/api/auth/logout", headers=headers)).json() == {"code": 0, "data": None, "message": ""}
+            denied = await client.get("/api/auth/me", headers=headers)
+            assert denied.status_code == 401 and denied.headers["www-authenticate"] == "Bearer"
+            assert (await client.get("/api/auth/me", headers={"Authorization": f"Bearer {second.json()['data']['token']}"})).status_code == 200
+    asyncio.run(check())
+
+
+def test_password_login_with_captcha_disabled(auth_app, monkeypatch):
+    from alphapulse.core.auth import get_settings
+    from alphapulse.core.config import Settings
+    app, memory = auth_app
+    assert Settings.model_fields["login_captcha_enabled"].default is False
+    monkeypatch.setattr(get_settings(), "login_captcha_enabled", False)
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            assert await challenge(client) == {"enabled": False}
+            assert memory.values == {}
+            body = {"username": "test-admin", "password": "test-password"}
+            signed = await client.post("/api/auth/login", json=body)
+            assert signed.status_code == 200
+            headers = {"Authorization": "Bearer " + signed.json()["data"]["token"]}
+            assert (await client.get("/api/auth/me", headers=headers)).json()["data"]["is_admin"] is True
+            assert (await client.post("/api/auth/logout", headers=headers)).status_code == 200
+            assert (await client.get("/api/auth/me", headers=headers)).status_code == 401
+            for _ in range(5):
+                assert (await client.post("/api/auth/login", json={**body, "password": "wrong"})).status_code == 401
+            locked = await client.post("/api/auth/login", json=body)
+            assert locked.status_code == 429 and int(locked.headers["retry-after"]) > 0
+            memory.offset += 901
+            assert (await client.post("/api/auth/login", json=body)).status_code == 200
+            memory.failed = True
+            assert (await client.post("/api/auth/login", json=body)).status_code == 503
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("fields", [{}, {"captcha_id": "challenge"}, {"code": "ABCDE"}])
+def test_enabled_captcha_cannot_be_omitted(auth_app, fields):
+    app, memory = auth_app
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            response = await client.post("/api/auth/login", json={"username": "test-admin", "password": "test-password", **fields})
+            assert response.status_code == 400 and response.json()["code"] == 400
+            assert not any("session:" in key for key in memory.values)
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("case", ["wrong", "expired", "replay", "concurrent"])
+def test_captcha_is_bound_consumed_and_expires(auth_app, case):
+    from alphapulse.api.auth import captcha_digest
+    from alphapulse.core.auth import AUTH_PREFIX
+    app, memory = auth_app
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            data = await challenge(client)
+            key = f"{AUTH_PREFIX}captcha:{data['captcha_id']}"
+            await memory.set(key, captcha_digest(data["captcha_id"], "ABCDE"), ex=120)
+            body = {"username": "test-admin", "password": "test-password", "captcha_id": data["captcha_id"], "code": "ABCDE"}
+            if case == "wrong":
+                body["code"] = "XYZ23"
+                assert (await client.post("/api/auth/login", json=body)).status_code == 400
+                body["code"] = "ABCDE"
+            if case == "expired":
+                memory.offset += 121
+            if case == "replay":
+                assert (await client.post("/api/auth/login", json=body)).status_code == 200
+            if case == "concurrent":
+                results = await asyncio.gather(*(client.post("/api/auth/login", json=body) for _ in range(2)))
+                assert sorted(result.status_code for result in results) == [200, 400]
+            else:
+                assert (await client.post("/api/auth/login", json=body)).status_code == 400
+            assert await memory.get(key) is None
+            body["captcha_id"] = "not-the-same-challenge"
+            assert (await client.post("/api/auth/login", json=body)).status_code == 400
+    asyncio.run(check())
+
+
+def test_login_limits_and_challenge_limits(auth_app):
+    app, memory = auth_app
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            for attempt in range(5):
+                if attempt: memory.offset += 100
+                assert (await login(client, memory, password="wrong")).status_code == 401
+            result = await login(client, memory)
+            assert result.status_code == 429 and int(result.headers["retry-after"]) >= 899
+            memory.offset += 901
+            assert (await login(client, memory)).status_code == 200
+            memory.offset += 61
+            for _ in range(30):
+                assert (await client.get("/api/auth/captcha")).status_code == 200
+            limited = await client.get("/api/auth/captcha")
+            assert limited.status_code == 429 and limited.headers["retry-after"]
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("case", ["cookie", "legacy", "expired", "forbidden", "redis"])
+def test_auth_failure_boundaries(auth_app, case):
+    from alphapulse.core import auth
+    app, memory = auth_app
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            token, _ = await auth.create_token("former-admin" if case == "forbidden" else "test-admin")
+            if case == "legacy":
+                token = jwt.encode({"sub": "test-admin", "exp": int(time.time()) + 60}, auth.get_settings().jwt_secret, algorithm="HS256")
+            if case == "expired":
+                payload = jwt.decode(token, auth.get_settings().jwt_secret, algorithms=["HS256"])
+                payload["exp"] = int(time.time()) - 1
+                token = jwt.encode(payload, auth.get_settings().jwt_secret, algorithm="HS256")
+            headers = {"Authorization": f"Bearer {token}"} if case != "cookie" else {"Cookie": f"ap_token={token}"}
+            memory.failed = case == "redis"
+            result = await client.get("/api/news", headers=headers)
+            assert result.status_code == (403 if case == "forbidden" else 503 if case == "redis" else 401)
+            assert result.json()["code"] == result.status_code
+            if case == "forbidden":
+                assert (await client.get("/api/auth/me", headers=headers)).json()["data"]["is_admin"] is False
+            if case == "redis":
+                assert (await client.get("/api/auth/captcha")).status_code == 503
+    asyncio.run(check())
+
+
+def test_protocol_status_validation_redaction_openapi(auth_app, monkeypatch):
+    from alphapulse.api import market
+    from alphapulse.core.auth import create_token
+    from alphapulse.core.db import get_session
+    app, memory = auth_app
+    app.dependency_overrides[get_session] = lambda: SimpleNamespace(add=lambda row: None, commit=AsyncMock())
+    monkeypatch.setattr(market, "tradable_symbols", AsyncMock(return_value=[{"symbol": "BTCUSDT", "base": "BTC", "quote": "USDT", "price_precision": 2, "volume_precision": 5}]))
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            assert (await client.get("/api/health")).json() == {"code": 0, "data": {"status": "ok"}, "message": ""}
+            invalid = await client.post("/api/auth/login", json={"password": {"sensitive": "do-not-echo"}})
+            assert invalid.status_code == 422 and invalid.json()["code"] == 422
+            assert "do-not-echo" not in invalid.text and "input" not in invalid.text
+            assert all(set(item) == {"loc", "type", "msg"} for item in invalid.json()["data"]["errors"])
+            assert (await client.get("/api/unknown")).json()["code"] == 404
+            token, _ = await create_token("test-admin")
+            client.headers["Authorization"] = f"Bearer {token}"
+            response = await client.post("/api/market/symbols", json={"market": "spot", "symbol": "BTCUSDT"})
+            assert response.status_code == 201 and response.json() == {"code": 0, "data": None, "message": ""}
+            schema = await client.get("/api/openapi.json")
+            assert "openapi" in schema.json() and "code" not in schema.json()
+            validation_schema = schema.json()["paths"]["/api/auth/login"]["post"]["responses"]["422"]["content"]["application/json"]["schema"]
+            assert "ApiResponse_ValidationErrors_" in validation_schema["$ref"]
+            assert schema.json()["paths"]["/api/auth/me"]["get"]["security"] == [{"HTTPBearer": []}]
+            assert (await client.get("/api/docs")).headers["location"] == "/docs"
+            monkeypatch.setattr(market, "tradable_symbols", AsyncMock(side_effect=RuntimeError("private provider failure")))
+            async with httpx.AsyncClient(transport=httpx.ASGITransport(app, raise_app_exceptions=False), base_url="http://test",
+                                         headers={"Authorization": f"Bearer {token}"}) as failing:
+                failed = await failing.post("/api/market/symbols", json={"market": "spot", "symbol": "BTCUSDT"})
+                assert failed.status_code == 500
+                assert failed.json() == {"code": 500, "data": None, "message": "服务器内部错误"}
+                assert "private provider failure" not in failed.text
+
+    asyncio.run(check())
+
+
+def test_stream_revoke_closes_producer_and_redis_failure_is_not_401(auth_app):
+    from alphapulse.core.auth import AuthSession, create_token, revoke_session
+    from alphapulse.core.streams import session_events
+    from fastapi.sse import ServerSentEvent
+    app, memory = auth_app
+    async def check():
+        closed = []
+        async def source():
+            try:
+                yield ServerSentEvent(data={"value": 1}, event="news")
+                await asyncio.Event().wait()
+            finally:
+                closed.append(True)
+        async def principal():
+            token, _ = await create_token("test-admin")
+            payload = jwt.decode(token, options={"verify_signature": False})
+            return AuthSession(payload["sub"], payload["jti"], payload["exp"])
+        session = await principal()
+        stream = session_events(source(), session)
+        assert (await anext(stream)).event == "news"
+        await revoke_session(session)
+        event = await asyncio.wait_for(anext(stream), timeout=2)
+        assert event.event == "auth_error" and event.data["code"] == 401
+        await stream.aclose()
+        assert closed == [True]
+        session = await principal()
+        memory.failed = True
+        stream = session_events(source(), session)
+        event = await anext(stream)
+        assert event.event == "error" and event.data["code"] == 503
+        await stream.aclose()
+    asyncio.run(check())
+
+
+def test_websocket_tickets_single_use_active_session(auth_app, monkeypatch):
+    from alphapulse.api import market
+    from alphapulse.core.auth import WS_PROTOCOL, create_token
+    app, memory = auth_app
+    monkeypatch.setattr(market.redis, "pubsub", lambda: SimpleNamespace(aclose=AsyncMock()))
+    token, _ = asyncio.run(create_token("test-admin"))
+    with TestClient(app) as client:
+        client.headers["Authorization"] = f"Bearer {token}"
+        expired = client.post("/api/auth/ws-ticket").json()["data"]["ticket"]
+        memory.offset += 31
+        with pytest.raises(WebSocketDenialResponse) as expired_response:
+            with client.websocket_connect("/api/market/ws", subprotocols=[WS_PROTOCOL, f"ticket.{expired}"]):
+                pass
+        assert expired_response.value.status_code == 401
+        ticket = client.post("/api/auth/ws-ticket").json()["data"]["ticket"]
+        unused = client.post("/api/auth/ws-ticket").json()["data"]["ticket"]
+        protocols = [WS_PROTOCOL, f"ticket.{ticket}"]
+        with client.websocket_connect("/api/market/ws", subprotocols=protocols) as socket:
+            assert socket.accepted_subprotocol == WS_PROTOCOL
+            client.post("/api/auth/logout")
+            assert socket.receive()["code"] == 4401
+        with pytest.raises(WebSocketDenialResponse) as denied:
+            with client.websocket_connect("/api/market/ws", subprotocols=protocols):
+                pass
+        assert denied.value.status_code == 401
+        with pytest.raises(WebSocketDenialResponse) as revoked:
+            with client.websocket_connect("/api/market/ws", subprotocols=[WS_PROTOCOL, f"ticket.{unused}"]):
+                pass
+        assert revoked.value.status_code == 401
+
+
+def test_trading_config_returns_single_envelope_and_manual_request_keeps_202(auth_app, monkeypatch):
+    from alphapulse.api import trading
+    from alphapulse.core.auth import create_token
+    from alphapulse.core.db import get_session
+    from alphapulse.models import TradingAccount
+    app, memory = auth_app
+    account = SimpleNamespace(exchange="bybit", provider_id=None, reasoning_effort=None, daily_analysis_limit=60)
+    async def get(model, key):
+        return account if model is TradingAccount else None
+    db = SimpleNamespace(get=get, add=lambda row: None, commit=AsyncMock(),
+                         execute=AsyncMock(return_value=SimpleNamespace(one=lambda: (2, 10, 5))))
+    app.dependency_overrides[get_session] = lambda: db
+    monkeypatch.setattr(trading, "read_snapshot", AsyncMock(return_value=None))
+    monkeypatch.setattr(trading, "get_symbol", AsyncMock(return_value=SimpleNamespace(running=True)))
+    monkeypatch.setattr(trading, "fresh_snapshot", AsyncMock())
+    monkeypatch.setattr(trading, "active_exchange", AsyncMock(return_value="bybit"))
+    async def check():
+        token, _ = await create_token("test-admin")
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test",
+                                     headers={"Authorization": f"Bearer {token}"}) as client:
+            result = await client.patch("/api/trading/config", json={"daily_analysis_limit": 120})
+            assert result.status_code == 200
+            assert result.json()["data"]["today"]["limit"] == 120
+            assert "code" not in result.json()["data"]
+            result = await client.post("/api/trading/symbols/BTCUSDT/manual/cancel")
+            assert result.status_code == 202
+            assert result.json() == {"code": 0, "data": None, "message": ""}
+    asyncio.run(check())

backend/tests/test_calendar.py

@@ -0,0 +1,328 @@
+import asyncio
+from datetime import UTC, date, datetime, timedelta
+from pathlib import Path
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from alphapulse.calendar.parsers import bea_release_link, parse_fed_calendar, parse_ics, parse_values
+
+FIXTURES = Path(__file__).parent / "fixtures" / "calendar"
+
+
+def ics(summary="Consumer Price Index", start="DTSTART;TZID=US-Eastern:20261110T083000", status=""):
+    return f"BEGIN:VCALENDAR\r\nBEGIN:VEVENT\r\nUID:stable-id\r\nSUMMARY:{summary}\r\n{start}\r\n{status}\r\nEND:VEVENT\r\nEND:VCALENDAR"
+
+
+@pytest.mark.parametrize("date,expected", [("20261014T083000", 12), ("20261110T083000", 13)])
+def test_schedule_respects_eastern_dst(date, expected):
+    [event] = parse_ics(ics(start=f"DTSTART;TZID=US-Eastern:{date}"), "bls")
+    assert event.at.hour == expected
+    assert event.external_id == "stable-id"
+
+
+def test_schedule_unfolding_utc_and_cancellation():
+    [event] = parse_ics(ics(summary="Personal Income and Outlays\\,\r\n September 2026",
+                            start="DTSTART:20261029T123000Z", status="STATUS:CANCELLED"), "bea")
+    assert event.family == "pce" and event.status == "withdrawn"
+    assert event.title == "Personal Income and Outlays,September 2026"
+    assert event.at == datetime(2026, 10, 29, 12, 30, tzinfo=UTC)
+
+
+@pytest.mark.parametrize("body,source", [("<html>Access denied</html>", "bea"),
+                                        (ics(start="DTSTART:20261110T083000"), "bls"),
+                                        (ics(summary="GDP by County and Personal Income by County, 2025"), "bea")])
+def test_invalid_or_irrelevant_schedule_is_not_empty_success(body, source):
+    with pytest.raises(ValueError):
+        parse_ics(body, source)
+
+
+def test_bea_discovery_matches_year_date_and_national_report():
+    def row(title, link):
+        return f'<tr class="scheduled-releases-type-press"><td class="release-date">September 30</td><td class="release-title"><a href="{link}">{title}</a></td></tr>'
+    body = '<table>' + row("GDP, (Third Estimate)", "/news/2026/gdp-third") + row("GDP by County", "/news/2026/county") + row("Personal Income and Outlays, August", "/news/2026/pce") + row("GDP, (Third Estimate)", "/news/2025/old") + '</table>'
+    assert bea_release_link(body, date(2026, 9, 30), "gdp") == "https://www.bea.gov/news/2026/gdp-third"
+    assert bea_release_link(body, date(2026, 9, 30), "pce") == "https://www.bea.gov/news/2026/pce"
+    assert bea_release_link(body, date(2026, 10, 29), "pce") is None
+    with pytest.raises(ValueError, match="缺少数据行"):
+        bea_release_link("<html>error</html>", date(2026, 9, 30), "pce")
+    with pytest.raises(ValueError, match="不唯一"):
+        bea_release_link(body + body, date(2026, 9, 30), "pce")
+
+
+def test_fed_calendar_uses_published_time_not_press_conference():
+    [event] = parse_fed_calendar((FIXTURES / "fed-month.html").read_text())
+    assert event.at == datetime(2026, 10, 28, 18, tzinfo=UTC)
+    assert event.family == "fomc"
+
+
+@pytest.mark.parametrize("family,key,actual,previous", [
+    ("cpi", "cpi_mom", "0.6", "0.4"), ("cpi", "core_cpi_yoy", "2.6", None),
+    ("ppi", "ppi_yoy", "6.3", "5.6"), ("jobs", "payrolls", "2.9", "13.3"),
+    ("jobs", "unemployment", "4.2", None), ("pce", "core_pce_mom", "0.2", "0.1"),
+    ("pce", "pce_yoy", "3.4", None), ("pce", "core_pce_yoy", "3.0", None),
+    ("fomc", "fed_rate", "4", None),
+])
+def test_official_release_values_and_revised_previous(family, key, actual, previous):
+    values = parse_values(family, (FIXTURES / f"{family}.html").read_text())
+    assert values[key] == {"actual": actual, "previous": previous}
+
+
+@pytest.mark.parametrize("rate,expected", [("4-1/2", "4.5"), ("1/4", "0.25"), ("4.25", "4.25")])
+def test_fed_fractional_upper_bound(rate, expected):
+    html = f'<div id="article">maintain the target range for the federal funds rate at 4 to {rate} percent</div>'
+    assert parse_values("fomc", html)["fed_rate"]["actual"] == expected
+
+
+def test_employment_decrease_and_negative_gdp():
+    html = '<div id="bodytext">Total nonfarm payroll employment declined by 92,000. The unemployment rate rose to 4.4 percent.</div>'
+    assert parse_values("jobs", html)["payrolls"]["actual"] == "-9.2"
+    html = '<div class="field--name-body">Real gross domestic product (GDP) decreased at an annual rate of 0.6 percent.</div>'
+    assert parse_values("gdp", html)["gdp"]["actual"] == "-0.6"
+
+
+def test_missing_release_value_is_not_zero():
+    with pytest.raises((ValueError, KeyError)):
+        parse_values("cpi", '<div id="bodytext">Release pending</div>')
+
+
+def test_poll_cadence_is_fast_at_release_and_bounded_after_errors(trading):
+    from alphapulse.calendar.service import next_poll
+    at = datetime(2026, 10, 29, 12, 30, tzinfo=UTC)
+    assert next_poll(at, at - timedelta(seconds=10), False) == at
+    assert next_poll(at, at, False) == at + timedelta(seconds=5)
+    assert next_poll(at, at + timedelta(minutes=5), False) == at + timedelta(minutes=5, seconds=15)
+    assert next_poll(at, at, False, True) == at + timedelta(seconds=60)
+    assert next_poll(at, at, True) == at + timedelta(seconds=60)
+
+
+def test_without_consensus_labels_reference_impact_without_claiming_surprise(trading):
+    from alphapulse.calendar.service import impact
+    direction, reason = impact("cpi_mom", "0.6", "0.4")
+    assert direction == "bearish" and "上升" in reason and "未取得市场一致预期" in reason
+    assert "参考判断" in reason and "不是超预期判断" in reason and "不代表币价必然" in reason
+    assert impact("cpi_mom", None, "0.4") == ("pending", "等待官方公布")
+
+
+@pytest.mark.parametrize("key,actual,previous,forecast,direction", [
+    ("cpi_mom", "0.2", "0.4", None, "bullish"),
+    ("ppi_mom", "0.2", "0.2", None, "neutral"),
+    ("fed_rate", "4.5", "4.25", None, "bearish"),
+    ("cpi_mom", "0.3", "0.2", "0.4", "bullish"),
+    ("payrolls", "2.9", "13.3", None, "uncertain"),
+    ("core_cpi_yoy", "2.6", None, None, "uncertain"),
+])
+def test_impact_uses_forecast_first_and_does_not_hide_ambiguous_channels(trading, key, actual, previous, forecast, direction):
+    from alphapulse.calendar.service import impact
+    actual_direction, explanation = impact(key, actual, previous, forecast)
+    assert actual_direction == direction
+    if forecast is not None:
+        assert "市场预期" in explanation and "不是超预期" not in explanation
+
+
+def test_event_has_country_specific_importance_and_comparison_basis(trading):
+    from alphapulse.calendar.service import release_rows
+    now = datetime(2026, 10, 24, tzinfo=UTC)
+    row = SimpleNamespace(id=1, period=None, stage=None, family="ppi", source="bls", title="PPI", scheduled_at=now,
+                          schedule_status="scheduled", first_received_at=now, values={"ppi_mom": {"actual": "0.2", "previous": "0.1"}},
+                          last_checked_at=now, error=None, release_url=None, schedule_url="https://www.bls.gov", revision_no=2)
+    [mom, _] = release_rows(row, now)
+    assert mom["country_code"] == "US" and mom["event_name"] == "美国PPI 最终需求月率"
+    assert mom["importance"] == 3 and mom["impact"] == "bearish" and mom["impact_basis"] == "previous"
+    row.family = "cpi"
+    assert release_rows(row, now)[1]["importance"] == 5
+
+
+def test_result_records_first_arrival_and_revisions_without_overwriting_history(trading):
+    from alphapulse.calendar.service import save_result
+    row = SimpleNamespace(id=1, period=None, stage=None, title="CPI", scheduled_at=datetime(2026, 10, 14, 12, 30, tzinfo=UTC),
+                          schedule_status="scheduled", values={}, release_url=None, first_received_at=None, revision_no=1)
+    captured = []
+    session = SimpleNamespace(add=captured.append)
+    now = row.scheduled_at + timedelta(seconds=7)
+    first = {"cpi_mom": {"actual": "0.6", "previous": "0.4"}}
+    save_result(session, row, first, "https://www.bls.gov/example", 1, now)
+    save_result(session, row, first, "https://www.bls.gov/example", 2, now + timedelta(seconds=5))
+    save_result(session, row, {"cpi_mom": {"actual": "0.5", "previous": "0.4"}}, "https://www.bls.gov/example", 3, now + timedelta(days=1))
+    assert row.first_received_at == now and row.revision_no == 3
+    assert len(captured) == 2 and captured[0].payload["values"] == first
+
+
+def test_result_date_mismatch_waits_without_using_stale_release(trading):
+    import httpx
+    from alphapulse.calendar.worker import CalendarWorker
+    async def check():
+        worker = CalendarWorker()
+        try:
+            response = httpx.Response(200, text='<html><body>October 14, 2026</body></html>',
+                                      request=httpx.Request("GET", "https://www.bls.gov/example"))
+            worker.fetch = AsyncMock(return_value=(response, 1))
+            row = SimpleNamespace(source="bls", family="cpi", scheduled_at=datetime(2026, 11, 10, 13, 30, tzinfo=UTC))
+            assert await worker.result_page(row) is None
+            row.scheduled_at = datetime(2026, 9, 10, 12, 30, tzinfo=UTC)
+            with pytest.raises(ValueError, match="日期与日历不符"):
+                await worker.result_page(row)
+        finally:
+            await worker.http.aclose()
+            await worker.bls.aclose()
+    asyncio.run(check())
+
+
+def test_schedule_reschedule_and_withdrawal_preserve_past_and_revisions(trading):
+    from alphapulse.calendar.parsers import Scheduled
+    from alphapulse.calendar.service import save_schedule
+    now = datetime(2026, 10, 24, tzinfo=UTC)
+    def row(id_, at):
+        return SimpleNamespace(id=id_, period=None, stage=None, external_id=str(id_), scheduled_at=at, title="CPI",
+                               schedule_status="scheduled", first_received_at=None, values={}, release_url=None,
+                               revision_no=1, schedule_checked_at=now, next_check_at=at)
+    rescheduled = row(1, now + timedelta(days=2))
+    removed = row(2, now + timedelta(days=3))
+    past = row(3, now - timedelta(days=3))
+    recorded = []
+    session = SimpleNamespace(scalars=AsyncMock(return_value=[rescheduled, removed, past]), add=recorded.append)
+    later = now + timedelta(days=4)
+    async def check():
+        events = [Scheduled("1", "cpi", "CPI", later)]
+        await save_schedule(session, "bls", "https://www.bls.gov/calendar", events, 9, now)
+        await save_schedule(session, "bls", "https://www.bls.gov/calendar", events, 10, now)
+    asyncio.run(check())
+    assert rescheduled.scheduled_at == later and rescheduled.next_check_at == later
+    assert removed.schedule_status == "withdrawn" and past.schedule_status == "scheduled"
+    assert len(recorded) == 2
+
+
+def test_calendar_api_requires_auth_validates_range_and_exposes_unavailable_worker(trading, monkeypatch):
+    import httpx
+    from fastapi import FastAPI
+    from alphapulse.api import calendar
+    from alphapulse.core.auth import require_admin, AuthSession
+    from alphapulse.core.config import get_settings
+    from alphapulse.core.db import get_session
+    monkeypatch.setattr(get_settings(), "jwt_secret", "calendar-review-test-only-32-bytes")
+    session = SimpleNamespace(scalars=AsyncMock(return_value=[]), get=AsyncMock(return_value=None))
+    app = FastAPI()
+    app.include_router(calendar.router)
+    app.dependency_overrides[get_session] = lambda: session
+    monkeypatch.setattr(calendar.redis, "get", AsyncMock(return_value=None))
+    async def check():
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app), base_url="http://test") as client:
+            assert (await client.get("/api/calendar?start=2026-10-24")).status_code == 401
+            assert (await client.get("/api/calendar/1/series?indicator=cpi_mom")).status_code == 401
+            app.dependency_overrides[require_admin] = lambda: AuthSession(get_settings().admin_username, "calendar-test", 9999999999)
+            response = await client.get("/api/calendar?start=2026-10-24")
+            assert response.status_code == 200
+            assert response.json()["data"]["worker"] is None and response.json()["data"]["forecast_available"] is False
+            assert response.json()["data"]["countries"] == [{"code": "US", "name": "美国"}, {"code": "EA", "name": "欧元区"}, {"code": "GB", "name": "英国"}, {"code": "DE", "name": "德国"}, {"code": "FR", "name": "法国"}]
+            assert (await client.get("/api/calendar?start=2026-10-24&days=32")).status_code == 422
+            assert (await client.get("/api/calendar/999/revisions")).status_code == 404
+            assert (await client.get("/api/calendar/999/history?indicator=cpi_mom")).status_code == 404
+            assert (await client.get("/api/calendar/999/series?indicator=cpi_mom")).status_code == 404
+            session.get.return_value = SimpleNamespace(family="cpi")
+            assert (await client.get("/api/calendar/1/history?indicator=ppi_mom")).status_code == 400
+            assert (await client.get("/api/calendar/1/history?indicator=cpi_mom&limit=121")).status_code == 422
+            assert (await client.get("/api/calendar/1/series?indicator=ppi_mom")).status_code == 400
+            assert (await client.get("/api/calendar/1/series?indicator=cpi_mom&limit=121")).status_code == 422
+            session.get.return_value = SimpleNamespace(family="cpi", period=None)
+            assert (await client.get("/api/calendar/1/series?indicator=cpi_mom")).json()["data"] == []
+            handler = AsyncMock(return_value=[{"actual": "0.3", "first_recorded_actual": "0.4", "revised": True}])
+            monkeypatch.setattr(calendar, "indicator_history", handler)
+            history = await client.get("/api/calendar/1/history?indicator=cpi_mom")
+            assert history.status_code == 200 and history.json()["data"][0]["revised"] is True
+            assert handler.await_args.args[2:] == ("cpi_mom", 36)
+    asyncio.run(check())
+
+
+def test_worker_initializes_all_sources_and_drops_expired_month_status(trading):
+    import httpx
+    from alphapulse.calendar.worker import CalendarWorker
+    async def check():
+        worker = CalendarWorker()
+        worker.sources["https://www.federalreserve.gov/old-month.htm"] = {"error": "old"}
+        async def unavailable(source, url, *, query=None):
+            assert len(worker.sources) == 15
+            assert all(s["error"] for s in worker.sources.values())
+            raise httpx.ConnectError("source unavailable")
+        worker.fetch = unavailable
+        try:
+            await worker.schedules()
+            assert all(s["error"] == "source unavailable" for s in worker.sources.values())
+            assert sum(s["source"] == "fed" for s in worker.sources.values()) == 2
+        finally:
+            await worker.http.aclose()
+            await worker.bls.aclose()
+    asyncio.run(check())
+
+
+@pytest.mark.parametrize("value,seconds", [("7200", 7200), ("Sun, 25 Oct 2026 14:00:00 GMT", 7200),
+                                           ("Sun, 25 Oct 2026 11:00:00 GMT", -3600),
+                                           (None, None), ("invalid", None), ("-1", None),
+                                           ("25 Oct 2026 14:00:00", None), ("9" * 30, None)])
+def test_retry_after_accepts_seconds_and_http_date_without_guessing_invalid_headers(trading, value, seconds):
+    from alphapulse.calendar.worker import retry_after_at
+    now = datetime(2026, 10, 25, 12, tzinfo=UTC)
+    assert retry_after_at(value, now) == (now + timedelta(seconds=seconds) if seconds is not None else None)
+
+
+@pytest.mark.parametrize("status", [429, 503])
+def test_limited_result_preserves_response_and_defers_single_poll(trading, monkeypatch, status):
+    import httpx
+    from alphapulse.calendar import worker as module
+
+    async def check():
+        now = datetime.now(UTC)
+        row = SimpleNamespace(id=1, source="eurostat", family="ea_hicp", scheduled_at=now,
+                              first_received_at=None, release_url="https://ec.europa.eu/eurostat/product?code=2-16102026-ap")
+        session = SimpleNamespace(scalar=AsyncMock(return_value=row), commit=AsyncMock())
+        class Context:
+            async def __aenter__(self):
+                return session
+            async def __aexit__(self, *args):
+                pass
+        monkeypatch.setattr(module, "SessionLocal", Context)
+        save_raw = AsyncMock(return_value=1)
+        monkeypatch.setattr(module, "save_raw", save_raw)
+        calls = []
+        def limited(request):
+            calls.append(request)
+            return httpx.Response(status, headers={"Retry-After": "7200"}, text="Quota exhausted")
+        worker = module.CalendarWorker()
+        await worker.http.aclose()
+        worker.http = httpx.AsyncClient(transport=httpx.MockTransport(limited))
+        try:
+            await worker.poll_one(1)
+            assert len(calls) == 1 and str(status) in row.error
+            assert row.first_received_at is None
+            assert row.next_check_at >= now + timedelta(seconds=7200)
+            assert save_raw.await_args.args[3] == status
+            assert save_raw.await_args.args[4]["retry-after"] == "7200"
+        finally:
+            await worker.http.aclose()
+            await worker.bls.aclose()
+    asyncio.run(check())
+
+
+def test_schedule_honors_server_pause_without_blocking_other_sources(trading):
+    import httpx
+    from alphapulse.calendar.worker import CalendarWorker
+    async def check():
+        worker = CalendarWorker()
+        calls = []
+        async def unavailable(source, url, *, query=None):
+            calls.append(source)
+            if source == "ecb":
+                httpx.Response(429, headers={"Retry-After": "7200"}, request=httpx.Request("GET", url)).raise_for_status()
+            raise httpx.ConnectError("source unavailable")
+        worker.fetch = unavailable
+        try:
+            await worker.schedules()
+            await worker.schedules()
+            assert calls.count("ecb") == 1
+            assert calls.count("insee") == 2
+            ecb = next(s for s in worker.sources.values() if s["source"] == "ecb")
+            assert ecb["error"] and datetime.fromisoformat(ecb["retry_at"]) > datetime.now(UTC)
+        finally:
+            await worker.http.aclose()
+            await worker.bls.aclose()
+    asyncio.run(check())

backend/tests/test_failure_paths.py

@@ -150,7 +150,7 @@ def test_chat_completes_tool_results_on_disconnect_or_exception(trading, monkeyp
     monkeypatch.setattr(chat.llm, "stream", stream)
     monkeypatch.setattr(chat, "run_tool", tool)
     async def check():
-        generator = chat.send_message(chat.SendIn(text="test"))
+        generator = chat._message_events(chat.SendIn(text="test"))
         events = []
         async for event in generator:
             events.append(event.event)
@@ -201,7 +201,7 @@ def test_chat_reset_blocks_new_generation_until_commit(trading, monkeypatch):
             task = asyncio.create_task(chat.reset(chat.ResetIn(clear_context=True), Session()))
             try:
                 await entered.wait()
-                events = [event async for event in chat.send_message(chat.SendIn(text="重置期间的新消息"))]
+                events = [event async for event in chat._message_events(chat.SendIn(text="重置期间的新消息"))]
                 assert len(events) == 1 and events[0].event == "error"
                 assert not task.done()
             finally:

backend/tests/test_news_sources.py

@@ -0,0 +1,318 @@
+import asyncio
+import json
+import re
+import uuid
+from datetime import UTC, datetime, timedelta
+from types import SimpleNamespace
+
+import pytest
+from scrapy.http import Request, XmlResponse
+from sqlalchemy import create_engine, select, text
+
+
+def make_spider(spider_type):
+    from alphapulse.crawler.spiders.base import RunState
+    spider = spider_type()
+    now = datetime(2026, 10, 21, tzinfo=UTC)
+    spider.run = RunState(uuid.uuid4(), "schedule", now, now - timedelta(hours=72))
+    return spider
+
+
+def feed_response(items, url="https://example.invalid/feed.xml"):
+    return XmlResponse(
+        url=url,
+        body=f'''<rss version="2.0" xmlns:dc="http://purl.org/dc/elements/1.1/"><channel>{items}</channel></rss>'''.encode(),
+        encoding="utf-8", request=Request(url, meta={"raw_response_id": 12}),
+    )
+
+
+ITEM = """<item><title>Bitcoin &amp; Ethereum</title>
+<link>https://example.invalid/news/bitcoin</link><guid>story-1</guid>
+<pubDate>Tue, 20 Oct 2026 08:30:00 EDT</pubDate>
+<description><![CDATA[<p>公开摘要 &amp; 数据</p>]]></description>
+<dc:creator>News desk</dc:creator><category>Markets</category></item>"""
+
+
+@pytest.mark.parametrize("source", ["CoinDeskSpider", "FedSpider", "BeaSpider"])
+def test_rss_preserves_original_summary_time_and_provenance(trading, source):
+    from alphapulse.crawler.spiders import rss
+    spider = make_spider(getattr(rss, source))
+    requests = list(spider.first_requests())
+    assert [request.url for request in requests] == list(spider.feeds)
+    assert all(request.meta["run_id"] == spider.run.id for request in requests)
+    item, = spider.parse_feed(feed_response(ITEM))
+    assert item["source_code"] == spider.name
+    assert item["title"] == "Bitcoin & Ethereum"
+    assert item["summary"] == item["content_text"] == "公开摘要 & 数据"
+    assert item["content_html"] == "<p>公开摘要 &amp; 数据</p>"
+    assert item["author"] == "News desk" and item["tags"] == ["Markets"]
+    assert item["published_at"] == datetime(2026, 10, 20, 12, 30, tzinfo=UTC)
+    assert item["source_updated_at"] is None
+    assert item["extra"] == {"language": "en", "content_scope": "feed_summary", "feed_url": "https://example.invalid/feed.xml"}
+    assert item["raw_response_id"] == 12
+
+
+def test_rss_identity_survives_revisions_and_different_subfeeds(trading):
+    from alphapulse.crawler.spiders.rss import FedSpider
+    spider = make_spider(FedSpider)
+    original, = spider.parse_feed(feed_response(ITEM))
+    revised, = spider.parse_feed(feed_response(ITEM.replace("Bitcoin &amp; Ethereum", "Revised headline"), "https://example.invalid/other.xml"))
+    assert original["external_id"] == revised["external_id"]
+    long_url = "https://example.invalid/" + "article" * 30
+    no_guid = ITEM.replace("<guid>story-1</guid>", "<guid> </guid>").replace("https://example.invalid/news/bitcoin", long_url)
+    first, = spider.parse_feed(feed_response(no_guid))
+    again, = spider.parse_feed(feed_response(no_guid))
+    assert len(first["external_id"]) == 64 and first["external_id"] == again["external_id"]
+    assert first["url"] == long_url
+
+
+@pytest.mark.parametrize("bad_item", [
+    ITEM.replace("Tue, 20 Oct 2026 08:30:00 EDT", "invalid date"),
+    ITEM.replace(" EDT", ""),
+    ITEM.replace("<title>Bitcoin &amp; Ethereum</title>", ""),
+    ITEM.replace("https://example.invalid/news/bitcoin", "javascript:alert(1)"),
+    ITEM.replace("<pubDate>Tue, 20 Oct 2026 08:30:00 EDT</pubDate>", ""),
+])
+def test_rss_isolates_bad_entries_and_records_visible_failure(trading, bad_item):
+    from alphapulse.crawler.spiders.rss import CoinDeskSpider
+    spider = make_spider(CoinDeskSpider)
+    items = list(spider.parse_feed(feed_response(bad_item + ITEM)))
+    assert len(items) == 1 and items[0]["title"] == "Bitcoin & Ethereum"
+    assert len(spider.run.errors) == 1 and "RSS 条目解析失败" in spider.run.errors[0]
+
+
+def test_rss_rejects_html_instead_of_reporting_empty_success(trading):
+    from alphapulse.crawler.spiders.rss import CoinDeskSpider
+    response = XmlResponse(url="https://example.invalid/rss", body=b"<html><body>Access denied</body></html>")
+    with pytest.raises(ValueError, match="不是 RSS/Atom"):
+        list(make_spider(CoinDeskSpider).parse_feed(response))
+
+
+@pytest.mark.parametrize("query,sources,expected", [
+    ("比特币", None, [1, 2, 16]),
+    ("BITCOIN", None, [1, 2, 16]),
+    ("BTC", ["coindesk"], [2, 16]),
+    ("以太坊", None, [3, 15]),
+    ("ETH", None, [3, 15]),
+    ("ether", None, [3, 15]),
+    ("美联储", None, [4, 5, 17]),
+    ("FOMC", None, [4, 5, 17]),
+    ("Federal Open Market Committee", ["fed"], [5, 17]),
+    ("CPI", None, [6, 7]),
+    ("非农", None, [8]),
+    ("GDP", None, [9]),
+    ("PCE", None, [10]),
+    ("50%_profit", None, [11]),
+    ("比特币 ETF", None, []),
+])
+def test_news_search_matches_topics_across_languages_and_keeps_literal_queries(trading, query, sources, expected):
+    from alphapulse.api.news import news_filters
+    from alphapulse.models import NewsItem
+    engine = create_engine("sqlite://")
+    try:
+        with engine.begin() as connection:
+            connection.execute(text("CREATE TABLE news_items (id INTEGER, source_code TEXT, title TEXT, content_text TEXT)"))
+            connection.execute(text("INSERT INTO news_items VALUES (:id, :source, :title, :body)"), [
+                {"id": i, "source": source, "title": title, "body": body}
+                for i, source, title, body in [
+                    (1, "cls_telegraph", "比特币消息", ""), (2, "coindesk", "Bitcoin market update", ""),
+                    (3, "coindesk", "Network update", "Ethereum update"),
+                    (4, "wscn_live", "美联储声明", ""), (5, "fed", "Federal Reserve statement", ""),
+                    (6, "bls", "Consumer Price Index", ""), (7, "cls_telegraph", "消费者价格指数公布", ""),
+                    (8, "bls", "Employment Situation", ""), (9, "bea", "Gross Domestic Product", ""),
+                    (10, "bea", "Personal Income and Outlays", "Personal consumption expenditures"),
+                    (11, "coindesk", "50%_profit", ""), (12, "coindesk", "50xxx_profit", ""),
+                    (13, "coindesk", "New method of employment research", ""), (14, "coindesk", "Ethernet hardware", ""),
+                    (15, "coindesk", "ETH update", ""), (16, "coindesk", "BTC news", ""),
+                    (17, "fed", "Minutes of the Federal Open Market Committee, September 15-16, 2026", ""),
+                ]
+            ])
+            statement = select(NewsItem.id).where(*news_filters(sources, query, None, None)).order_by(NewsItem.id)
+            assert connection.execute(statement).scalars().all() == expected
+    finally:
+        engine.dispose()
+
+
+@pytest.mark.parametrize("headline,matched", [
+    ("Bitcoin ETF inflows", True), ("ethereum network update", True),
+    ("ETH rally", True), ("CRYPTO regulation", True), ("Stablecoins update", True),
+    ("Binance announcement", True), ("比特币消息", True), ("微策略增持", True),
+    ("New method for employment statistics", False), ("Ethernet hardware demand", False),
+    ("A hypothetical company", False),
+])
+def test_english_crypto_filter_uses_whole_words(trading, headline, matched):
+    from alphapulse.trading.news_signals import CRYPTO_PATTERN
+    assert bool(re.search(CRYPTO_PATTERN, headline, re.IGNORECASE)) is matched
+
+
+def test_macro_label_receives_summary_separately_from_truncated_title(trading, monkeypatch):
+    from alphapulse.ai import llm
+    from alphapulse.models import TradingAccount
+    from alphapulse.trading import news_signals
+
+    title = "GDP, Industries, Corporate Profits, State GDP, and State Personal Income; " * 4
+    summary = "Real gross domestic product increased at an annual rate of 2.2 percent."
+    provider = SimpleNamespace(model="test-only")
+    writes = []
+
+    class Session:
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *args):
+            pass
+        async def execute(self, statement):
+            if statement.is_select:
+                return SimpleNamespace(all=lambda: [(123, title, summary)])
+            writes.append(statement.compile().params)
+        async def get(self, model, key):
+            return SimpleNamespace(provider_id=1) if model is TradingAccount else provider
+        def expunge(self, value):
+            assert value is provider
+        async def commit(self):
+            pass
+
+    async def stream(actual_provider, system, messages, tools, effort):
+        assert actual_provider is provider
+        item, = json.loads(messages[0]["text"])
+        assert item["id"] == 123 and item["summary"] == summary
+        assert len(item["title"]) == 200
+        assert "neutral" in system and "市场预期" in system
+        yield llm.Completion(tool_calls=[{"name": "label", "arguments": {
+            "items": [{"id": 123, "category": "event", "direction": "neutral"}],
+        }}])
+
+    monkeypatch.setattr(news_signals, "SessionLocal", Session)
+    monkeypatch.setattr(llm, "stream", stream)
+    assert asyncio.run(news_signals.label_new()) == 1
+    assert writes[0]["news_id_m0"] == 123 and writes[0]["direction_m0"] == "neutral"
+
+
+def test_news_tools_accept_foreign_sources_and_disclose_summary_scope(trading, monkeypatch):
+    from alphapulse.ai import tools
+    args = tools.SearchNews(query="CPI", sources=["coindesk", "fed", "bls", "bea"])
+    assert len(args.sources) == 4
+    news = SimpleNamespace(
+        id=1, source_code="fed", published_at=datetime(2026, 10, 21, tzinfo=UTC),
+        title="Policy statement", url="https://example.invalid/statement", is_paywalled=False,
+        content_text="Public feed summary", extra={"content_scope": "feed_summary"},
+    )
+    class Session:
+        async def __aenter__(self):
+            return self
+        async def __aexit__(self, *args):
+            pass
+        async def get(self, model, news_id):
+            return news
+    monkeypatch.setattr(tools, "SessionLocal", Session)
+    result = json.loads(asyncio.run(tools.get_news(tools.GetNews(id=1))).result)
+    assert result["source"] == "美联储·政策与讲话"
+    assert result["content_scope"] == "feed_summary" and result["content"] == news.content_text
+
+
+@pytest.mark.parametrize("bad_date", [None, "2026-10-20T08:30:00", "invalid date", ""])
+def test_atom_uses_bls_published_time_and_isolates_invalid_dates(trading, bad_date):
+    from alphapulse.crawler.spiders.rss import BlsSpider
+    published = bad_date if bad_date is not None else "2026-10-20T08:30:00.2-04:00"
+    entry = f"""<entry><id>cpi-release</id><title>Consumer Price Index</title>
+    <link href="https://www.bls.gov/news.release/archives/cpi_10202026.htm"/><published>{published}</published>
+    <updated>2026-10-20T09:00:00-04:00</updated>
+    <content>Monthly change &amp; value &lt; 1%</content><category>News Release</category></entry>"""
+    url = "https://example.invalid/cpi.rss"
+    response = XmlResponse(url=url, body=f'<feed xmlns="http://www.w3.org/2005/Atom">{entry}</feed>'.encode(),
+                           request=Request(url, meta={"raw_response_id": 13}))
+    spider = make_spider(BlsSpider)
+    items = list(spider.parse_feed(response))
+    if bad_date is not None:
+        assert not items and len(spider.run.errors) == 1
+    else:
+        item, = items
+        assert item["published_at"] == datetime(2026, 10, 20, 12, 30, tzinfo=UTC)
+        assert item["extra"]["feed_published_at"] == "2026-10-20T12:30:00.200000+00:00"
+        assert item["source_updated_at"] == datetime(2026, 10, 20, 13, tzinfo=UTC)
+        assert item["summary"] == item["content_text"] == "Monthly change & value < 1%"
+        assert "&lt;" in item["content_html"] and item["tags"] == ["News Release"]
+        assert item["extra"]["content_scope"] == "feed_summary" and not spider.run.errors
+
+
+@pytest.mark.parametrize("program,date,feed_time,expected", [
+    ("cpi", "10142026", "2026-10-14T07:50:31.2-04:00", datetime(2026, 10, 14, 12, 30, tzinfo=UTC)),
+    ("ppi", "10152026", "2026-10-15T07:50:51.146-04:00", datetime(2026, 10, 15, 12, 30, tzinfo=UTC)),
+    ("empsit", "10022026", "2026-10-02T07:51:08.289-04:00", datetime(2026, 10, 2, 12, 30, tzinfo=UTC)),
+    ("cpi", "01132026", "2026-01-13T07:50:31-05:00", datetime(2026, 1, 13, 13, 30, tzinfo=UTC)),
+])
+def test_bls_uses_official_release_time_and_keeps_feed_timestamp(trading, program, date, feed_time, expected):
+    from alphapulse.crawler.spiders.rss import BlsSpider
+    from alphapulse.trading.news_signals import is_forward_sample
+
+    url = f"https://www.bls.gov/feed/{program}.rss"
+    response = XmlResponse(url=url, body=f'''<feed xmlns="http://www.w3.org/2005/Atom"><entry>
+        <id>{program}-{date}</id><title>Official release</title>
+        <link href="https://www.bls.gov/news.release/archives/{program}_{date}.htm"/>
+        <published>{feed_time}</published><content>Public summary</content>
+        </entry></feed>'''.encode(), request=Request(url, meta={"raw_response_id": 13}))
+    spider = make_spider(BlsSpider)
+    item, = spider.parse_feed(response)
+    assert item["published_at"] == expected and not spider.run.errors
+    assert datetime.fromisoformat(item["extra"]["feed_published_at"]) == datetime.fromisoformat(feed_time)
+    assert item["extra"]["published_at_basis"] == "bls_release_schedule"
+    assert is_forward_sample(item["published_at"], expected + timedelta(minutes=1))
+    assert not is_forward_sample(item["published_at"], expected + timedelta(minutes=6))
+
+
+def test_bls_rejects_unrecognized_release_date_instead_of_guessing(trading):
+    from alphapulse.crawler.spiders.rss import BlsSpider
+
+    spider = make_spider(BlsSpider)
+    assert not list(spider.parse_feed(feed_response(ITEM)))
+    assert len(spider.run.errors) == 1 and "公告日期" in spider.run.errors[0]
+
+
+def test_bea_repairs_only_its_observed_official_schemeless_link(trading):
+    from alphapulse.crawler.spiders.rss import BeaSpider, CoinDeskSpider
+    response = feed_response(ITEM.replace("https://example.invalid/news/bitcoin", "www.bea.gov/news/2026/gdp"))
+    item, = make_spider(BeaSpider).parse_feed(response)
+    assert item["url"] == "https://www.bea.gov/news/2026/gdp"
+    other = make_spider(CoinDeskSpider)
+    assert not list(other.parse_feed(response)) and len(other.run.errors) == 1
+
+
+@pytest.mark.parametrize("status", [200, 302, 403])
+def test_bls_downloader_preserves_compressed_body_status_and_redirect_for_scrapy(trading, monkeypatch, status):
+    import gzip
+    import ssl
+
+    import httpx
+
+    from alphapulse.crawler.spiders import rss
+    original_client = httpx.AsyncClient
+    body = gzip.compress(b"<feed>Public announcement</feed>")
+    def handle(request):
+        assert request.method == "GET"
+        assert request.headers["accept-language"] == "en-US"
+        return httpx.Response(status, headers={"Content-Encoding": "gzip", "Content-Type": "application/atom+xml",
+                                               "Location": "https://www.bls.gov/feed/cpi.rss"}, stream=httpx.ByteStream(body))
+    def client(**kwargs):
+        context = kwargs["verify"]
+        assert context.check_hostname and context.verify_mode == ssl.CERT_REQUIRED
+        assert context.minimum_version >= ssl.TLSVersion.TLSv1_2 and context.security_level >= 2
+        return original_client(transport=httpx.MockTransport(handle), **kwargs)
+    monkeypatch.setattr(rss.httpx, "AsyncClient", client)
+    request = Request("https://www.bls.gov/feed/cpi.rss", headers={"Accept-Language": "en-US"})
+    response = asyncio.run(rss.BlsDownloadHandler(15).download_request(request))
+    assert response.status == status and response.body == body
+    assert response.headers["Content-Encoding"] == b"gzip"
+    assert response.headers["Location"] == b"https://www.bls.gov/feed/cpi.rss"
+    assert response.request is request
+
+
+def test_bls_transport_error_reaches_existing_scrapy_retry_boundary(trading, monkeypatch):
+    import httpx
+
+    from alphapulse.crawler.spiders import rss
+    original_client = httpx.AsyncClient
+    def handle(request):
+        raise httpx.ReadTimeout("transient timeout", request=request)
+    monkeypatch.setattr(rss.httpx, "AsyncClient", lambda **kwargs: original_client(transport=httpx.MockTransport(handle), **kwargs))
+    with pytest.raises(httpx.ReadTimeout, match="transient timeout") as exc:
+        asyncio.run(rss.BlsDownloadHandler(15).download_request(Request("https://www.bls.gov/feed/cpi.rss")))
+    assert isinstance(exc.value, tuple(x for x in rss.BlsSpider.custom_settings["RETRY_EXCEPTIONS"] if isinstance(x, type)))

backend/tests/test_research_accounting.py

@@ -19,7 +19,7 @@ def test_forward_endpoint_excludes_labels_at_or_after_entry(trading):
     class Session:
         async def execute(self, statement):
             return SimpleNamespace(all=lambda: rows)
-    result = asyncio.run(list_news_signals(Session()))
+    result = asyncio.run(list_news_signals(Session())).data
     assert result.labeled == result.filled == 1
     assert [item.news_id for item in result.recent] == [1]
     assert all(s.n == 1 for s in result.stats)