mirror of
https://github.com/OpenByteInc/QuantDinger.git
synced 2026-09-28 15:22:56 +08:00
461 lines
19 KiB
Python
461 lines
19 KiB
Python
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from app.professional_report.builder import build_professional_report
|
|
from app.professional_report.llm_contract import validate_llm_analysis
|
|
from app.professional_report.prompt import build_professional_analysis_prompt
|
|
from app.professional_report.risk import build_risk_plan
|
|
from app.professional_report.snapshot import build_evidence_snapshot
|
|
|
|
|
|
def _collector_payload(market="USStock"):
|
|
now = datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
|
|
symbol = {"USStock": "AAPL", "HKStock": "00700", "Crypto": "ETH/USDT@swap"}[market]
|
|
payload = {
|
|
"market": market,
|
|
"symbol": symbol,
|
|
"timeframe": "1D",
|
|
"collected_at": now,
|
|
"price": {"price": 100, "changePercent": 1.2, "source": "test_quote"},
|
|
"kline": [{"timestamp": now, "open": 98, "high": 102, "low": 97, "close": 100, "volume": 1000}],
|
|
"indicators": {
|
|
"rsi": {"value": 57, "signal": "neutral"},
|
|
"moving_averages": {"trend": "uptrend"},
|
|
"macd": {"signal": "bullish"},
|
|
"levels": {"support": 95, "resistance": 110},
|
|
"trading_levels": {
|
|
"suggested_stop_loss": 95,
|
|
"suggested_take_profit": 110,
|
|
},
|
|
"volatility": {"atr": 3, "pct": 3},
|
|
},
|
|
"news": [{"title": "Confirmed product update", "source": "wire", "published_at": now}],
|
|
"_meta": {"success_items": ["price", "kline", "indicators"], "failed_items": [], "duration_ms": 12},
|
|
}
|
|
if market in {"USStock", "HKStock"}:
|
|
payload["fundamental"] = {
|
|
"source": "test_fundamental",
|
|
"market_cap": 1_000_000_000,
|
|
"revenue_growth": 12.5,
|
|
"profit_margin": 18.0,
|
|
"pe_ratio": 20,
|
|
"field_metadata": {
|
|
key: {"source": "test_fundamental", "as_of": now, "unit": unit}
|
|
for key, unit in {
|
|
"market_cap": "USD" if market == "USStock" else "HKD",
|
|
"revenue_growth": "percent",
|
|
"profit_margin": "percent",
|
|
"pe_ratio": "ratio",
|
|
}.items()
|
|
},
|
|
"financial_statements": {"latest_quarter": {"period_end": now[:10]}},
|
|
"identity": {"verified": True, "reported_symbol": symbol},
|
|
}
|
|
else:
|
|
payload["crypto_instrument"] = {"venue": "binance", "market_type": "perpetual"}
|
|
payload["crypto_factors"] = {
|
|
"volume_24h": 500_000_000,
|
|
"funding_rate": 0.01,
|
|
"funding_rate_decimal": 0.0001,
|
|
"open_interest": 250_000_000,
|
|
"open_interest_change_24h": 2.5,
|
|
"long_short_ratio": 1.1,
|
|
"sources": {"market_structure": "coingecko", "derivatives": "binance_public"},
|
|
"metric_metadata": {
|
|
"volume_24h": {"unit": "usd", "currency": "USD", "provider": "coingecko", "venue": "aggregate", "product_type": "spot"},
|
|
"funding_rate": {"unit": "percent", "provider": "binance_public", "venue": "binance", "product_type": "perpetual"},
|
|
"open_interest": {"unit": "usd", "currency": "USD", "provider": "binance_public", "venue": "binance", "product_type": "perpetual"},
|
|
"open_interest_change_24h": {"unit": "percent", "provider": "binance_public", "venue": "binance", "product_type": "perpetual"},
|
|
"long_short_ratio": {"unit": "ratio", "provider": "binance_public", "venue": "binance", "product_type": "perpetual"},
|
|
},
|
|
}
|
|
return payload
|
|
|
|
|
|
def _analysis(payload, evidence_claims=None):
|
|
return {
|
|
"market": payload["market"],
|
|
"language": "zh-CN",
|
|
"decision": "BUY",
|
|
"confidence": 88,
|
|
"summary": "趋势改善\u00e2\u20ac\u201d但仍需确认\ufffd",
|
|
"timeframe": "medium",
|
|
"detailed_analysis": {"technical": "动量改善", "fundamental": "数据可用", "sentiment": "中性"},
|
|
"scores": {"technical": 68, "fundamental": 60, "sentiment": 52},
|
|
"objective_score": {"macro_score": 5},
|
|
"consensus": {"consensus_score": 24},
|
|
"market_data": {"current_price": 100, "support": 95, "resistance": 110},
|
|
"trading_plan": {"entry_price": 100, "stop_loss": 95, "take_profit": 110, "position_size_pct": 20},
|
|
"reasons": ["技术趋势改善"],
|
|
"risks": ["跌破支撑的风险"],
|
|
"evidence_claims": evidence_claims or [],
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("market", ["USStock", "HKStock", "Crypto"])
|
|
def test_professional_builder_produces_valid_contract_for_supported_markets(market):
|
|
payload = _collector_payload(market)
|
|
snapshot = build_evidence_snapshot(payload)
|
|
ref = snapshot["observations"][0]["evidence_id"]
|
|
report = build_professional_report(
|
|
payload,
|
|
_analysis(payload, [{"kind": "thesis", "text": "当前价格证据可追溯", "evidence_refs": [ref]}]),
|
|
)
|
|
|
|
assert report["contract_validation"]["valid"] is True
|
|
assert report["instrument"]["market"] == market
|
|
assert report["claims"][0]["evidence_refs"] == [ref]
|
|
assert "\u00e2\u20ac\u201d" not in report["executive_summary"]
|
|
assert "\ufffd" not in report["executive_summary"]
|
|
assert report["data_quality"]["coverage_ratio"] == 1
|
|
|
|
|
|
def test_missing_required_equity_data_blocks_directional_recommendation():
|
|
payload = _collector_payload("USStock")
|
|
payload["fundamental"] = {}
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
|
|
assert report["decision_profile"]["raw_decision"] == "BUY"
|
|
assert report["decision_profile"]["decision"] == "HOLD"
|
|
assert report["decision_profile"]["confidence"] <= 35
|
|
assert set(report["data_quality"]["missing_metrics"]) >= {
|
|
"market_cap", "revenue_growth", "profit_margin"
|
|
}
|
|
|
|
|
|
def test_financial_periods_are_separate_evidence_observations():
|
|
payload = _collector_payload("USStock")
|
|
payload["fundamental"]["financial_statements"] = {
|
|
"latest_quarter": {
|
|
"period_end": "2026-06-30",
|
|
"income_statement": {"total_revenue": 100},
|
|
},
|
|
"ttm": {
|
|
"period_end": "2026-06-30",
|
|
"income_statement": {"total_revenue": 390},
|
|
},
|
|
"latest_annual": {
|
|
"period_end": "2025-12-31",
|
|
"income_statement": {"total_revenue": 350},
|
|
},
|
|
}
|
|
snapshot = build_evidence_snapshot(payload)
|
|
by_metric = {item["metric"]: item for item in snapshot["observations"]}
|
|
|
|
assert by_metric["financial.latest_quarter.income_statement.total_revenue"]["value"] == 100
|
|
assert by_metric["financial.ttm.income_statement.total_revenue"]["value"] == 390
|
|
assert by_metric["financial.latest_annual.income_statement.total_revenue"]["value"] == 350
|
|
assert by_metric["financial.latest_quarter.income_statement.total_revenue"]["period_end"] == "2026-06-30"
|
|
|
|
|
|
def test_risk_plan_is_cost_and_quality_aware():
|
|
plan = build_risk_plan(
|
|
"BUY",
|
|
100,
|
|
{"entry_price": 100, "stop_loss": 95, "take_profit": 110, "position_size_pct": 80},
|
|
data_quality_score=50,
|
|
market="USStock",
|
|
account_risk_budget_pct=1,
|
|
estimated_roundtrip_cost_bps=20,
|
|
)
|
|
|
|
assert plan["valid"] is True
|
|
assert plan["net_risk_reward"] < plan["gross_risk_reward"]
|
|
assert plan["recommended_position_pct"] <= 25
|
|
assert "position_reduced_for_data_quality" in plan["warnings"]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("technical_score", "expected_bias", "expected_direction"),
|
|
[(-16, "BEARISH", "SELL"), (8, "BULLISH", "BUY")],
|
|
)
|
|
def test_hold_keeps_market_bias_and_non_actionable_candidate_geometry(
|
|
technical_score, expected_bias, expected_direction
|
|
):
|
|
payload = _collector_payload("USStock")
|
|
analysis = _analysis(payload)
|
|
analysis["decision"] = "HOLD"
|
|
analysis["objective_score"]["technical_score"] = technical_score
|
|
analysis["trading_plan"] = {
|
|
"entry_price": 0,
|
|
"stop_loss": 0,
|
|
"take_profit": 0,
|
|
}
|
|
|
|
report = build_professional_report(payload, analysis)
|
|
profile = report["decision_profile"]
|
|
plan = report["risk_plan"]
|
|
candidate = plan["candidate_setup"]
|
|
|
|
assert report["contract_validation"]["valid"] is True
|
|
assert profile["decision"] == "HOLD"
|
|
assert profile["market_bias"] == expected_bias
|
|
assert profile["market_bias_score"] == technical_score
|
|
assert plan["entry_price"] is None
|
|
assert plan["recommended_position_pct"] == 0
|
|
assert plan["max_position_pct"] == 0
|
|
assert candidate["status"] == "watch_only"
|
|
assert candidate["direction"] == expected_direction
|
|
assert candidate["entry_price"] == 100
|
|
assert candidate["net_risk_reward"] > 0
|
|
|
|
|
|
def test_neutral_hold_does_not_invent_candidate_geometry():
|
|
payload = _collector_payload("Crypto")
|
|
analysis = _analysis(payload)
|
|
analysis["decision"] = "HOLD"
|
|
analysis["objective_score"]["technical_score"] = 2
|
|
|
|
report = build_professional_report(payload, analysis)
|
|
|
|
assert report["decision_profile"]["market_bias"] == "NEUTRAL"
|
|
assert report["risk_plan"]["candidate_setup"] is None
|
|
|
|
|
|
def test_invalid_price_geometry_blocks_actionable_decision():
|
|
payload = _collector_payload("USStock")
|
|
analysis = _analysis(payload)
|
|
analysis["trading_plan"] = {
|
|
"entry_price": 100,
|
|
"stop_loss": 105,
|
|
"take_profit": 110,
|
|
"position_size_pct": 20,
|
|
}
|
|
report = build_professional_report(payload, analysis)
|
|
|
|
assert report["decision_profile"]["decision"] == "HOLD"
|
|
assert report["risk_plan"]["valid"] is False
|
|
assert "invalid_risk_plan" in report["decision_profile"]["quality_gate_reasons"]
|
|
|
|
|
|
def test_requested_professional_tier_downgrades_without_professional_evidence():
|
|
payload = _collector_payload("USStock")
|
|
report = build_professional_report(payload, _analysis(payload), data_tier="professional")
|
|
|
|
assert report["data_tier"] == "community"
|
|
assert report["methodology"]["requested_data_tier"] == "professional"
|
|
assert "professional_tier_requested_but_no_professional_evidence" in report["warnings"]
|
|
|
|
|
|
def test_macro_dimension_has_deterministic_narrative_and_evidence():
|
|
payload = _collector_payload("HKStock")
|
|
payload["macro"] = {
|
|
"DXY": {"price": 98.87, "change": 0, "changePercent": 0},
|
|
"FEAR_GREED": {"price": 71, "change": 0, "changePercent": 0},
|
|
}
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
macro = next(item for item in report["dimensions"] if item["key"] == "macro")
|
|
|
|
assert macro["status"] == "available"
|
|
assert "美元指数" in macro["narrative"]
|
|
assert "不应单独作为买卖依据" in macro["narrative"]
|
|
assert macro["evidence_refs"]
|
|
|
|
|
|
def test_macro_dimension_uses_market_specific_context():
|
|
payload = _collector_payload("USStock")
|
|
payload["macro"] = {"VIX": {"price": 16.2}}
|
|
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
macro = next(item for item in report["dimensions"] if item["key"] == "macro")
|
|
|
|
assert "美国风险资产环境" in macro["narrative"]
|
|
assert "香港资金环境" not in macro["narrative"]
|
|
|
|
|
|
def test_compact_provider_timestamp_is_normalized_before_contract_validation():
|
|
payload = _collector_payload("USStock")
|
|
payload["news"] = [{
|
|
"title": "Confirmed filing update",
|
|
"source": "GDELT",
|
|
"published_at": "20260907T230000Z",
|
|
}]
|
|
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
news_rows = [
|
|
item for item in report["evidence_snapshot"]["observations"]
|
|
if item["category"] == "news"
|
|
]
|
|
|
|
assert news_rows
|
|
assert news_rows[0]["as_of"] == "2026-09-07T23:00:00Z"
|
|
|
|
|
|
def test_future_dated_provider_evidence_is_excluded_without_failing_report():
|
|
payload = _collector_payload("USStock")
|
|
payload["collected_at"] = "2099-09-08T12:00:00Z"
|
|
payload["news"] = [
|
|
{
|
|
"title": "Valid confirmed update",
|
|
"source": "wire",
|
|
"published_at": "2099-09-08T11:55:00Z",
|
|
},
|
|
{
|
|
"title": "Provider timestamp in the future",
|
|
"source": "wire",
|
|
"published_at": "2099-09-08T20:00:00Z",
|
|
},
|
|
]
|
|
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
snapshot = report["evidence_snapshot"]
|
|
news_values = [
|
|
item["value"] for item in snapshot["observations"]
|
|
if item["category"] == "news"
|
|
]
|
|
|
|
assert [item["title"] for item in news_values] == ["Valid confirmed update"]
|
|
assert "future_timestamp_evidence_excluded" in snapshot["quality_flags"]
|
|
assert snapshot["collection"]["excluded_future_timestamp_items"] == 1
|
|
assert report["contract_validation"]["valid"] is True
|
|
|
|
|
|
def test_hk_dimension_uses_free_enrichment_and_skips_inapplicable_ah_premium():
|
|
payload = _collector_payload("HKStock")
|
|
payload["hk_security_profile"] = {
|
|
"security_type": "非H股",
|
|
"is_h_share": False,
|
|
"southbound_eligible_sh": True,
|
|
"source": "eastmoney_hk_via_akshare",
|
|
}
|
|
payload["southbound_flow"] = {
|
|
"holding_change_pct_1d": 1.25,
|
|
"scope": "stock_connect_holdings_change_proxy",
|
|
"source": "eastmoney_hsgt_via_akshare",
|
|
"source_url": "https://example.test/holdings",
|
|
"as_of": payload["collected_at"],
|
|
}
|
|
payload["analyst_expectations"] = {
|
|
"rating_direction": "买入",
|
|
"target_price_median_hkd": 520,
|
|
"source": "etnet_hk_via_akshare",
|
|
"as_of": payload["collected_at"],
|
|
}
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
dimension = next(item for item in report["dimensions"] if item["key"] == "market_specific")
|
|
|
|
assert "ah_premium" not in report["market_features"]["missing_capabilities"]
|
|
assert "南向持股一日变化" in dimension["narrative"]
|
|
assert "目标价中位数" in dimension["narrative"]
|
|
assert dimension["evidence_refs"]
|
|
southbound_evidence = next(
|
|
item for item in report["evidence_snapshot"]["observations"]
|
|
if item["metric"] == "southbound_flow.snapshot"
|
|
)
|
|
assert "source_url" not in southbound_evidence["value"]
|
|
assert southbound_evidence["source_url"] == "https://example.test/holdings"
|
|
|
|
|
|
def test_hk_holdings_proxy_cannot_be_reported_as_net_flow():
|
|
payload = _collector_payload("HKStock")
|
|
payload["southbound_flow"] = {
|
|
"holding_change_pct_1d": 1.25,
|
|
"scope": "stock_connect_holdings_change_proxy",
|
|
"source": "eastmoney_hsgt_via_akshare",
|
|
"as_of": payload["collected_at"],
|
|
}
|
|
analysis = _analysis(payload)
|
|
analysis["summary"] = "南向资金小幅净流入。"
|
|
analysis["detailed_analysis"]["sentiment"] = "南向资金净买入支持股价。"
|
|
report = build_professional_report(payload, analysis)
|
|
|
|
assert "净流入" not in report["executive_summary"]
|
|
assert "南向持股" in report["executive_summary"]
|
|
sentiment = next(item for item in report["dimensions"] if item["key"] == "news_sentiment")
|
|
assert "净买入" not in sentiment["narrative"]
|
|
assert "持仓变化代理" in sentiment["narrative"]
|
|
|
|
|
|
def test_us_market_dimension_describes_scope_instead_of_overclaiming():
|
|
payload = _collector_payload("USStock")
|
|
payload.update({
|
|
"sec_filings": [{"form": "10-Q", "filing_date": "2026-08-01", "source": "sec_edgar"}],
|
|
"analyst_expectations": {
|
|
"rating_direction": "buy", "analyst_count": 30,
|
|
"target_price_median_usd": 120, "source": "yahoo_finance",
|
|
},
|
|
"options": {
|
|
"expiry": "2026-09-18", "put_call_open_interest_ratio": 0.8,
|
|
"nearest_atm_implied_volatility_pct": 25, "scope": "nearest_expiry_snapshot",
|
|
"source": "yahoo_finance",
|
|
},
|
|
"short_interest": {
|
|
"short_percent_of_float_pct": 1.2, "short_ratio_days": 1.5,
|
|
"scope": "reported_short_interest_not_daily_short_volume", "source": "yahoo_finance",
|
|
},
|
|
"insider_activity": {
|
|
"recent_form4_filing_count": 4,
|
|
"scope": "form4_filing_activity_not_trade_direction", "source": "sec_edgar",
|
|
},
|
|
})
|
|
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
dimension = next(item for item in report["dimensions"] if item["key"] == "market_specific")
|
|
|
|
assert dimension["status"] == "available"
|
|
assert "SEC披露" in dimension["narrative"]
|
|
assert "最近到期期权快照" in dimension["narrative"]
|
|
assert "不表示买卖方向" in dimension["narrative"]
|
|
assert dimension["evidence_refs"]
|
|
|
|
|
|
def test_professional_crypto_source_marks_effective_professional_tier():
|
|
payload = _collector_payload("Crypto")
|
|
payload["crypto_factors"]["sources"]["derivatives"] = "coinglass"
|
|
for key in ("funding_rate", "open_interest", "open_interest_change_24h", "long_short_ratio"):
|
|
payload["crypto_factors"]["metric_metadata"][key]["provider"] = "coinglass"
|
|
report = build_professional_report(payload, _analysis(payload), data_tier="community")
|
|
|
|
assert report["data_tier"] == "professional"
|
|
|
|
|
|
def test_ambiguous_crypto_scope_blocks_directional_report():
|
|
payload = _collector_payload("Crypto")
|
|
payload["crypto_factors"]["metric_metadata"]["funding_rate"].pop("unit")
|
|
report = build_professional_report(payload, _analysis(payload))
|
|
|
|
assert report["decision_profile"]["decision"] == "HOLD"
|
|
assert report["decision_profile"]["confidence"] <= 35
|
|
assert "crypto_scope_or_unit_validation_failed" in report["decision_profile"]["quality_gate_reasons"]
|
|
|
|
|
|
def test_llm_contract_drops_unknown_and_ungrounded_claims():
|
|
fallback = {
|
|
"decision": "HOLD", "confidence": 35, "summary": "fallback",
|
|
"analysis": {"technical": "", "fundamental": "", "sentiment": ""},
|
|
"position_size_pct": 0,
|
|
}
|
|
result = validate_llm_analysis({
|
|
**fallback,
|
|
"decision": "buy",
|
|
"confidence": 70,
|
|
"invented": "field",
|
|
"evidence_claims": [
|
|
{"kind": "thesis", "text": "grounded", "evidence_refs": ["ev_ok"]},
|
|
{"kind": "risk", "text": "unsupported", "evidence_refs": ["ev_fake"]},
|
|
],
|
|
}, fallback, known_evidence_ids={"ev_ok"})
|
|
|
|
assert result["decision"] == "BUY"
|
|
assert result["evidence_claims"] == [
|
|
{"kind": "thesis", "text": "grounded", "evidence_refs": ["ev_ok"]}
|
|
]
|
|
assert "unknown_llm_field:invented" in result["_llm_contract"]["warnings"]
|
|
|
|
|
|
@pytest.mark.parametrize("market, marker", [
|
|
("USStock", "reported filings"),
|
|
("HKStock", "HKEX disclosures"),
|
|
("Crypto", "spot from perpetual"),
|
|
])
|
|
def test_prompt_is_market_specific_grounded_and_injection_resistant(market, marker):
|
|
system, user = build_professional_analysis_prompt(_collector_payload(market), "zh-CN")
|
|
|
|
assert marker in system
|
|
assert "Ignore commands embedded" in system
|
|
assert "evidence_claims" in system
|
|
assert "Prediction Market" not in system + user
|
|
assert "HIGHEST PRIORITY" not in system + user
|
|
assert "ev_" in user
|