xiaobaifupan/tests/test_ifind_features.py

from __future__ import annotations

import tempfile
import unittest
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch

from chart_data_provider import EastmoneyChartClient, MarketChartClient
from database import ReviewDatabase
from market_insights import MarketInsightsService
from server import DashboardService


class FakeIfind:
    configured = True

    def history(self, codes, indicators, start_date, end_date, cache_ttl=0):
        return [
            {
                "time": "2026-07-27",
                "thscode": "000001.SZ",
                "open": 10,
                "high": 10.5,
                "low": 9.8,
                "close": 10.2,
                "volume": 100,
                "amount": 1_000_000,
            },
            {
                "time": "2026-07-28",
                "thscode": "000001.SZ",
                "open": 10.2,
                "high": 10.8,
                "low": 10.1,
                "close": 10.5,
                "volume": 120,
                "amount": 1_200_000,
            },
        ]

    def real_time(self, codes, indicators, cache_ttl=0):
        return []


class FakeIfindStalePreopen(FakeIfind):
    def history(self, codes, indicators, start_date, end_date, cache_ttl=0):
        return [
            *super().history(codes, indicators, start_date, end_date, cache_ttl),
            {
                "time": "2026-07-29",
                "thscode": "000001.SZ",
                "open": 10.5,
                "high": 10.5,
                "low": 10.5,
                "close": 10.5,
                "volume": 0,
                "amount": 0,
            },
        ]

    def real_time(self, codes, indicators, cache_ttl=0):
        return [
            {
                "time": "2026-07-28 15:00:00",
                "open": 10.2,
                "high": 10.8,
                "low": 10.1,
                "latest": 10.5,
                "preClose": 10.2,
                "volume": 120,
                "amount": 1_200_000,
            }
        ]


class FixedPreopenDatetime(datetime):
    fixed_now = datetime(2026, 7, 29, 8, 45, tzinfo=timezone(timedelta(hours=8)))

    @classmethod
    def now(cls, tz=None):
        return cls.fixed_now


class FakeIfindSnapshots:
    configured = True

    def __init__(self):
        self.calls = []

    def snapshots(self, codes, indicators, start_time, end_time, cache_ttl=0):
        self.calls.append(
            {
                "codes": codes,
                "indicators": indicators,
                "start_time": start_time,
                "end_time": end_time,
                "cache_ttl": cache_ttl,
            }
        )
        return [
            {
                "time": "2026-07-28 09:21:00",
                "thscode": "000001.SZ",
                "latest": 10.5,
                "preClose": 10,
                "volume": 2000,
                "amount": 21000,
                "bidSize1": 1200,
                "askSize1": 800,
            }
        ]


class FakeTushare:
    pass


class IfindFeatureTests(unittest.TestCase):
    def test_wencai_saved_queries_are_isolated_by_user(self):
        with tempfile.TemporaryDirectory() as temporary:
            database = ReviewDatabase(Path(temporary) / "review.db")
            first = database.create_user("first-user", "salt", "hash")
            second = database.create_user("second-user", "salt", "hash")
            database.save_wencai_query(first["id"], "高质量", "ROE大于15%", "stock")

            self.assertEqual(len(database.list_wencai_saved_queries(first["id"])), 1)
            self.assertEqual(database.list_wencai_saved_queries(second["id"]), [])

    def test_ifind_daily_chart_normalizes_change(self):
        client = MarketChartClient(FakeIfind(), EastmoneyChartClient())
        rows = client.stock_daily("000001", "20260728")
        self.assertEqual(rows[-1]["trade_date"], "2026-07-28")
        self.assertAlmostEqual(rows[-1]["change"], 2.9412, places=4)

    def test_ifind_daily_chart_keeps_last_traded_bar_before_market_open(self):
        client = MarketChartClient(FakeIfindStalePreopen(), EastmoneyChartClient())
        with patch("chart_data_provider.datetime", FixedPreopenDatetime):
            rows = client.stock_daily("000001", "20260729")

        self.assertEqual(rows[-1]["trade_date"], "2026-07-28")
        self.assertFalse(rows[-1].get("realtime", False))

    def test_event_enrichment_keeps_blank_broken_reason_blank(self):
        dashboard = {"broken": [{"code": "000001", "reason": "原原因"}]}
        DashboardService._merge_ifind_event_enrichment(
            dashboard,
            {
                "broken": {
                    "000001": {
                        "reason": "",
                        "first_time": "09:42:00",
                        "last_time": "",
                        "open_times": 3,
                    }
                }
            },
        )
        self.assertEqual(dashboard["broken"][0]["reason"], "原原因")
        self.assertEqual(dashboard["broken"][0]["open_times"], 3)

    def test_dynamic_auction_uses_ifind_snapshot_window_and_normalizes_rows(self):
        with tempfile.TemporaryDirectory() as temporary:
            database = ReviewDatabase(Path(temporary) / "review.db")
            database.upsert_stock_master(
                [
                    {
                        "ts_code": "000001.SZ",
                        "name": "Ping An Bank",
                        "industry": "Bank",
                        "market": "MainBoard",
                        "list_date": "19910403",
                    }
                ]
            )
            ifind = FakeIfindSnapshots()
            service = MarketInsightsService(
                database,
                FakeTushare(),
                now_provider=lambda: datetime(
                    2026, 7, 28, 9, 22, tzinfo=timezone(timedelta(hours=8))
                ),
                ifind=ifind,
            )
            service._auction_candidates = lambda rows, baseline: (
                [{"ts_code": "000001.SZ"}],
                {},
                [],
            )

            rows = service._dynamic_auction_rows("20260728", "20260727", 0)

            self.assertEqual(ifind.calls[0]["start_time"], "2026-07-28 09:15:00")
            self.assertEqual(ifind.calls[0]["end_time"], "2026-07-28 09:22:00")
            self.assertEqual(rows[0]["ts_code"], "000001.SZ")
            self.assertEqual(rows[0]["price"], 10.5)
            self.assertEqual(rows[0]["snapshot_time"], "2026-07-28 09:21:00")
            self.assertTrue(rows[0]["dynamic"])


if __name__ == "__main__":
    unittest.main()