mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
09afbf566f
* fix(issue-1412): [bug]-stock_list格式问题 * fix(review-feedback-1413): preserve exchange hint for dotted A-share inputs * fix(review-feedback-1413): Keep normalized A-share codes usable by market routing and preserve * fix(review-feedback-1413): Limit raw dotted codes to fetchers that can parse them * fix(review-feedback-1413): Keep Tushare daily input normalized for ETF detection * fix(review-feedback-1413): 澄清结构化检测中的外部模型/API 与运行时配置迁移风险 * fix(review-feedback-1413): 处理或明确确认该失败与本 PR 无关且已有维护者豁免依据 * fix: keep stock list input as bare codes * docs: add phase-0 run diagnostics contract * fix(review-feedback-1435): 修正描述并澄清/补齐运行时代码变更的验证证据 * fix(review-feedback-1435): 补齐前缀提示识别，并增加对应回归测试 * fix(review-feedback-1435): 修正 * fix(review-feedback-1435): 解决冲突并更新描述/验证记录后再合入 * fix(review-feedback-1435): 解决冲突，并在最终 head 上重新确认 python -m pytest tests/test a share fetcher code * fix(review-feedback-1435): 修复并补齐回归覆盖后再复核最终 head * fix(review-feedback-1435): data provider/baostock fetcher.py 的 convert stock code 只从 .SH/.SS/.SZ * fix: preserve A-share exchange hints * fix(review-feedback-1435): 修正 docs/run-diagnostics-p0.md 对 Tushare 本轮范围的矛盾描述
125 lines
5.4 KiB
Python
125 lines
5.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Regression tests for provider-side A-share stock code conversion.
|
|
"""
|
|
|
|
import unittest
|
|
|
|
import pandas as pd
|
|
|
|
from data_provider.base import DataFetcherManager, normalize_stock_code
|
|
from data_provider.baostock_fetcher import BaostockFetcher
|
|
from data_provider.pytdx_fetcher import PytdxFetcher
|
|
from data_provider.tushare_fetcher import TushareFetcher
|
|
|
|
|
|
class _RecordingDailyFetcher:
|
|
name = "RecordingDailyFetcher"
|
|
priority = 1
|
|
|
|
def __init__(self) -> None:
|
|
self.calls = []
|
|
|
|
def get_daily_data(self, stock_code: str, *args, **kwargs) -> pd.DataFrame:
|
|
self.calls.append(stock_code)
|
|
return pd.DataFrame({"date": ["2026-05-22"], "close": [10.0]})
|
|
|
|
|
|
class TestDataFetcherManagerAShareCodes(unittest.TestCase):
|
|
def test_get_daily_data_keeps_user_contract_as_bare_stock_code(self) -> None:
|
|
fetcher = _RecordingDailyFetcher()
|
|
manager = DataFetcherManager(fetchers=[fetcher])
|
|
|
|
df, source = manager.get_daily_data("601888", days=1)
|
|
|
|
self.assertFalse(df.empty)
|
|
self.assertEqual(source, "RecordingDailyFetcher")
|
|
self.assertEqual(fetcher.calls, ["601888"])
|
|
|
|
|
|
class TestBaostockAShareCodeConversion(unittest.TestCase):
|
|
def test_convert_bare_stock_codes_to_baostock_format(self) -> None:
|
|
fetcher = BaostockFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("601888"), "sh.601888")
|
|
self.assertEqual(fetcher._convert_stock_code("600519"), "sh.600519")
|
|
self.assertEqual(fetcher._convert_stock_code("605499"), "sh.605499")
|
|
self.assertEqual(fetcher._convert_stock_code("688981"), "sh.688981")
|
|
self.assertEqual(fetcher._convert_stock_code("000001"), "sz.000001")
|
|
self.assertEqual(fetcher._convert_stock_code("001979"), "sz.001979")
|
|
self.assertEqual(fetcher._convert_stock_code("003816"), "sz.003816")
|
|
self.assertEqual(fetcher._convert_stock_code("300750"), "sz.300750")
|
|
self.assertEqual(fetcher._convert_stock_code("301012"), "sz.301012")
|
|
|
|
def test_convert_bare_etf_codes_to_baostock_format(self) -> None:
|
|
fetcher = BaostockFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("510050"), "sh.510050")
|
|
self.assertEqual(fetcher._convert_stock_code("159919"), "sz.159919")
|
|
|
|
def test_convert_suffix_code_uses_internal_provider_format(self) -> None:
|
|
fetcher = BaostockFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("600519.SH"), "sh.600519")
|
|
self.assertEqual(fetcher._convert_stock_code("000001.SZ"), "sz.000001")
|
|
|
|
def test_convert_prefix_code_preserves_explicit_exchange_hint(self) -> None:
|
|
fetcher = BaostockFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("SH000001"), "sh.000001")
|
|
self.assertEqual(fetcher._convert_stock_code("SH.000001"), "sh.000001")
|
|
self.assertEqual(fetcher._convert_stock_code("SZ600519"), "sz.600519")
|
|
self.assertEqual(fetcher._convert_stock_code("SZ.600519"), "sz.600519")
|
|
self.assertEqual(fetcher._convert_stock_code("ss.600519"), "sh.600519")
|
|
|
|
|
|
class TestPytdxAShareCodeConversion(unittest.TestCase):
|
|
def test_get_market_code_for_bare_stock_codes(self) -> None:
|
|
fetcher = PytdxFetcher(hosts=[])
|
|
|
|
self.assertEqual(fetcher._get_market_code("601888"), (1, "601888"))
|
|
self.assertEqual(fetcher._get_market_code("688981"), (1, "688981"))
|
|
self.assertEqual(fetcher._get_market_code("000001"), (0, "000001"))
|
|
self.assertEqual(fetcher._get_market_code("300750"), (0, "300750"))
|
|
|
|
def test_get_market_code_preserves_explicit_exchange_hint(self) -> None:
|
|
fetcher = PytdxFetcher(hosts=[])
|
|
|
|
self.assertEqual(fetcher._get_market_code("SH000001"), (1, "000001"))
|
|
self.assertEqual(fetcher._get_market_code("SH.000001"), (1, "000001"))
|
|
self.assertEqual(fetcher._get_market_code("SZ600519"), (0, "600519"))
|
|
self.assertEqual(fetcher._get_market_code("SZ.600519"), (0, "600519"))
|
|
self.assertEqual(fetcher._get_market_code("ss.600519"), (1, "600519"))
|
|
|
|
|
|
class TestTushareAShareCodeConversion(unittest.TestCase):
|
|
def test_convert_bare_stock_codes_to_tushare_format(self) -> None:
|
|
fetcher = TushareFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("605499"), "605499.SH")
|
|
self.assertEqual(fetcher._convert_stock_code("001979"), "001979.SZ")
|
|
self.assertEqual(fetcher._convert_stock_code("003816"), "003816.SZ")
|
|
self.assertEqual(fetcher._convert_stock_code("301012"), "301012.SZ")
|
|
|
|
def test_convert_prefix_code_preserves_explicit_exchange_hint(self) -> None:
|
|
fetcher = TushareFetcher()
|
|
|
|
self.assertEqual(fetcher._convert_stock_code("SH000001"), "000001.SH")
|
|
self.assertEqual(fetcher._convert_stock_code("SH.000001"), "000001.SH")
|
|
self.assertEqual(fetcher._convert_stock_code("SZ600519"), "600519.SZ")
|
|
self.assertEqual(fetcher._convert_stock_code("SZ.600519"), "600519.SZ")
|
|
self.assertEqual(fetcher._convert_stock_code("ss.600519"), "600519.SH")
|
|
|
|
|
|
class TestNormalizeStockCode(unittest.TestCase):
|
|
def test_normalize_prefixed_dot_code(self) -> None:
|
|
self.assertEqual(normalize_stock_code("SH.600519"), "600519")
|
|
self.assertEqual(normalize_stock_code("sh.600519"), "600519")
|
|
self.assertEqual(normalize_stock_code("SZ.000001"), "000001")
|
|
self.assertEqual(normalize_stock_code("sz.000001"), "000001")
|
|
self.assertEqual(normalize_stock_code("BJ.920748"), "920748")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|