229 lines
8.2 KiB
Python
229 lines
8.2 KiB
Python
from __future__ import annotations
|
|
|
|
import ast
|
|
import unittest
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
from tushare_client import _sector_coverage_issue
|
|
|
|
|
|
def load_method(name: str):
|
|
source = Path("backend/application.py").read_text(encoding="utf-8")
|
|
tree = ast.parse(source)
|
|
dashboard_service = next(
|
|
node for node in tree.body
|
|
if isinstance(node, ast.ClassDef) and node.name == "DashboardService"
|
|
)
|
|
method = next(
|
|
node for node in dashboard_service.body
|
|
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
|
|
and node.name == name
|
|
)
|
|
module = ast.Module(body=[method], type_ignores=[])
|
|
namespace = {
|
|
"datetime": datetime,
|
|
"Any": object,
|
|
"_sector_coverage_issue": _sector_coverage_issue,
|
|
}
|
|
exec(compile(ast.fix_missing_locations(module), "server.py", "exec"), namespace)
|
|
return namespace[name]
|
|
|
|
|
|
MARKET_MODE = load_method("_heaven_market_mode")
|
|
QUALITY_ISSUES = load_method("_heaven_trend_quality_issues")
|
|
TZ = timezone(timedelta(hours=8))
|
|
|
|
|
|
class MarketModeTests(unittest.TestCase):
|
|
def test_closed_rt_snapshot_is_closed_not_intraday(self):
|
|
dashboard = {"meta": {"realtime": True, "market_status": "closed"}}
|
|
now = datetime(2026, 7, 20, 16, 27, tzinfo=TZ)
|
|
self.assertEqual(MARKET_MODE("20260720", dashboard, now), "closed")
|
|
|
|
def test_trading_snapshot_is_intraday(self):
|
|
dashboard = {"meta": {"realtime": True, "market_status": "trading"}}
|
|
now = datetime(2026, 7, 20, 10, 30, tzinfo=TZ)
|
|
self.assertEqual(MARKET_MODE("20260720", dashboard, now), "intraday")
|
|
|
|
def test_historical_date_is_always_historical(self):
|
|
dashboard = {"meta": {"realtime": True, "market_status": "trading"}}
|
|
now = datetime(2026, 7, 20, 10, 30, tzinfo=TZ)
|
|
self.assertEqual(MARKET_MODE("20260717", dashboard, now), "historical")
|
|
|
|
@staticmethod
|
|
def intraday_layers(trade_date: str):
|
|
index_context = {
|
|
"trade_date": trade_date,
|
|
"precise": True,
|
|
"realtime": True,
|
|
"indices": [{"trade_date": trade_date}] * 3,
|
|
}
|
|
sector = {
|
|
"trade_date": trade_date,
|
|
"precise": True,
|
|
"realtime": True,
|
|
"taxonomy": "sw_l2",
|
|
"schema_version": 3,
|
|
"coverage": 100,
|
|
"relative_turnover": 1.2,
|
|
}
|
|
stock = {
|
|
"trade_date": trade_date,
|
|
"code": "002141",
|
|
"precise": True,
|
|
"realtime": True,
|
|
"turnover_source": "float_share",
|
|
"activity_source": "historical_progress",
|
|
}
|
|
return index_context, sector, stock
|
|
|
|
@staticmethod
|
|
def historical_layers(trade_date: str):
|
|
index_context = {
|
|
"trade_date": trade_date,
|
|
"precise": True,
|
|
"realtime": False,
|
|
"source": "tushare",
|
|
"indices": [{"trade_date": trade_date}] * 3,
|
|
}
|
|
sector = {
|
|
"trade_date": trade_date,
|
|
"precise": True,
|
|
"realtime": False,
|
|
"taxonomy": "sw_l2",
|
|
"schema_version": 3,
|
|
"source": "tushare_sw_daily+member_daily",
|
|
"coverage": 97,
|
|
}
|
|
stock = {
|
|
"trade_date": trade_date,
|
|
"code": "002141",
|
|
"precise": True,
|
|
"realtime": False,
|
|
"data_source": "tushare",
|
|
}
|
|
return index_context, sector, stock
|
|
|
|
def test_intraday_accepts_verified_realtime_layers(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.intraday_layers(trade_date)
|
|
dashboard = {
|
|
"meta": {
|
|
"realtime": True,
|
|
"market_status": "closed",
|
|
"updated_at": datetime.now(TZ).isoformat(),
|
|
}
|
|
}
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, dashboard, index_context, sector, stock, "intraday"
|
|
)
|
|
self.assertEqual(issues, [])
|
|
|
|
def test_intraday_sector_coverage_90_passes_89_blocks(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.intraday_layers(trade_date)
|
|
dashboard = {"meta": {"realtime": True, "market_status": "closed"}}
|
|
sector["coverage"] = 90
|
|
self.assertEqual(
|
|
QUALITY_ISSUES(trade_date, dashboard, index_context, sector, stock, "intraday"),
|
|
[],
|
|
)
|
|
sector["coverage"] = 89
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, dashboard, index_context, sector, stock, "intraday"
|
|
)
|
|
self.assertTrue(any("覆盖率" in issue for issue in issues))
|
|
|
|
def test_intraday_sector_requires_relative_turnover(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.intraday_layers(trade_date)
|
|
sector["relative_turnover"] = 0
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "intraday"
|
|
)
|
|
self.assertTrue(any("相对全市场换手" in issue for issue in issues))
|
|
|
|
def test_sector_must_use_shenwan_l2_taxonomy(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.intraday_layers(trade_date)
|
|
sector["taxonomy"] = "ths"
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "intraday"
|
|
)
|
|
self.assertTrue(any("申万二级" in issue for issue in issues))
|
|
|
|
def test_historical_accepts_official_daily_layers(self):
|
|
trade_date = "20260717"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "historical"
|
|
)
|
|
self.assertEqual(issues, [])
|
|
|
|
def test_historical_rejects_realtime_index_layer(self):
|
|
trade_date = "20260717"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
index_context["realtime"] = True
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "historical"
|
|
)
|
|
self.assertTrue(any("指数层" in issue for issue in issues))
|
|
|
|
def test_historical_rejects_realtime_sector_layer(self):
|
|
trade_date = "20260717"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
sector["realtime"] = True
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "historical"
|
|
)
|
|
self.assertTrue(any("行业层" in issue for issue in issues))
|
|
|
|
def test_historical_rejects_non_tushare_stock(self):
|
|
trade_date = "20260717"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
stock["data_source"] = "dashboard"
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "historical"
|
|
)
|
|
self.assertTrue(any("个股层" in issue for issue in issues))
|
|
|
|
def test_stock_must_be_precise(self):
|
|
trade_date = "20260717"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
stock["precise"] = False
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "historical"
|
|
)
|
|
self.assertTrue(any("个股层" in issue for issue in issues))
|
|
|
|
def test_closed_mode_ignores_nonessential_dashboard_status(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
dashboard = {"meta": {"realtime": True, "market_status": "trading"}}
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, dashboard, index_context, sector, stock, "closed"
|
|
)
|
|
self.assertEqual(issues, [])
|
|
|
|
def test_closed_mode_accepts_finalized_realtime_shenwan_snapshot(self):
|
|
trade_date = "20260720"
|
|
index_context, sector, stock = self.historical_layers(trade_date)
|
|
sector.update({
|
|
"realtime": True,
|
|
"finalized": True,
|
|
"inner_precise": True,
|
|
"outer_precise": True,
|
|
"relative_turnover": 1.2,
|
|
})
|
|
|
|
issues = QUALITY_ISSUES(
|
|
trade_date, {"meta": {}}, index_context, sector, stock, "closed"
|
|
)
|
|
|
|
self.assertEqual(issues, [])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|