web_development/backend/app/services/market_data.py

import asyncio
import logging
import random
from collections import defaultdict
from datetime import datetime, timedelta, timezone

import numpy as np

from app.core.contracts import (
    CATEGORIES,
    CONTRACTS,
    EXCHANGES,
    ContractInfo,
    get_contracts_by_category,
    get_contracts_by_exchange,
)
from app.models.schemas import KlineData, MarketData

logger = logging.getLogger(__name__)

_INTERVAL_MINUTES = {"1m": 1, "5m": 5, "15m": 15, "30m": 30, "1h": 60, "4h": 240, "1d": 1440, "1w": 10080}
_BAR_COUNT = {"1m": 500, "5m": 300, "15m": 200, "30m": 200, "1h": 200, "4h": 150, "1d": 120, "1w": 60}
_VOLATILITY = {"1m": 0.0004, "5m": 0.0009, "15m": 0.0015, "30m": 0.002, "1h": 0.003, "4h": 0.005, "1d": 0.01, "1w": 0.02}


def _now():
    return datetime.now(timezone.utc)


class MarketDataService:
    INTERVALS = list(_INTERVAL_MINUTES.keys())

    def __init__(self):
        self._quotes: dict[str, MarketData] = {}
        self._klines: dict[str, dict[str, list[KlineData]]] = defaultdict(lambda: defaultdict(list))
        self._subscribers: list[asyncio.Queue] = []
        self._running = False
        self._task: asyncio.Task | None = None
        self._mode: str = "simulated"
        self._real_data_cache: dict[str, dict] = {}
        self._sim_prices: dict[str, float] = {}
        self._init_simulated()

    @property
    def mode(self):
        return self._mode

    def set_mode(self, mode: str):
        if mode not in ("simulated", "realtime"):
            raise ValueError("mode must be 'simulated' or 'realtime'")
        self._mode = mode

    def get_exchanges(self):
        return EXCHANGES

    def get_categories(self):
        return CATEGORIES

    def get_contracts(self, category=None, exchange=None):
        cs = get_contracts_by_category(category) if category else get_contracts_by_exchange(exchange) if exchange else CONTRACTS
        return [{"symbol": c.symbol, "exchange": c.exchange, "category": c.category, "name_cn": c.name_cn} for c in cs]

    def get_current_price(self, symbol):
        q = self._quotes.get(symbol)
        return q.close if q else self._sim_prices.get(symbol)

    def get_quote(self, symbol):
        return self._quotes.get(symbol)

    def get_all_market_data(self):
        return list(self._quotes.values())

    def get_kline_history(self, symbol, interval="1m", limit=200):
        return self._klines.get(symbol, {}).get(interval, [])[-limit:]

    def subscribe(self):
        q: asyncio.Queue = asyncio.Queue(maxsize=200)
        self._subscribers.append(q)
        return q

    def unsubscribe(self, q):
        if q in self._subscribers:
            self._subscribers.remove(q)

    async def _notify(self, data):
        dead = []
        for q in self._subscribers:
            try:
                q.put_nowait(data)
            except asyncio.QueueFull:
                dead.append(q)
        for d in dead:
            self._subscribers.remove(d)

    async def start(self):
        if self._running:
            return
        self._running = True
        self._task = asyncio.create_task(self._run())
        logger.info("MarketDataService started (%s, %d contracts)", self._mode, len(CONTRACTS))

    async def stop(self):
        self._running = False
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass

    async def _run(self):
        tick = 0
        while self._running:
            try:
                if self._mode == "realtime" and tick % 5 == 0:
                    await self._fetch_realtime()
                else:
                    await self._tick_sim()
            except Exception as e:
                logger.error("Tick error: %s", e)
            tick += 1
            await asyncio.sleep(1)

    # ── Realtime ─────────────────────────────────────────────────────────────

    async def _fetch_realtime(self):
        loop = asyncio.get_event_loop()
        for c in CONTRACTS:
            try:
                df = await loop.run_in_executor(None, self._fetch_one, c.symbol)
                if df is not None and len(df) > 0:
                    r = df.iloc[0]
                    md = MarketData(
                        symbol=c.symbol, name=c.name_cn, exchange=c.exchange, category=c.category,
                        open=float(r.get("open", 0)), high=float(r.get("high", 0)),
                        low=float(r.get("low", 0)),
                        close=float(r.get("trade", 0) or r.get("close", 0)),
                        volume=int(r.get("volume", 0)),
                        bid=float(r.get("bidprice1", 0) or 0), ask=float(r.get("askprice1", 0) or 0),
                        open_interest=int(r.get("position", 0) or 0),
                        pre_close=float(r.get("preclose", 0) or 0),
                        pre_settlement=float(r.get("prevsettlement", 0) or r.get("presettlement", 0) or 0),
                        settlement=float(r.get("settlement", 0) or 0),
                        change_pct=float(r.get("changepercent", 0) or 0),
                        timestamp=_now(),
                    )
                    self._quotes[c.symbol] = md
                    await self._notify(md)
                    self._real_data_cache[c.symbol] = {"contracts": df.to_dict("records"), "updated_at": _now().isoformat()}
            except Exception as e:
                logger.debug("Fetch %s: %s", c.symbol, e)

    @staticmethod
    def _fetch_one(symbol):
        import akshare as ak
        return ak.futures_zh_realtime(symbol=symbol)

    def get_contract_details(self, symbol):
        return self._real_data_cache.get(symbol)

    # ── Simulated ────────────────────────────────────────────────────────────

    _BASE = {
        "沪深300指数期货": 3800, "上证50指数期货": 2500, "中证500指数期货": 5600,
        "中证1000股指期货": 5200, "5年期国债期货": 102.5, "10年期国债期货": 104.2,
        "2年期国债期货": 101.3,
        "黄金": 530, "白银": 7200, "铂": 980, "钯": 1050,
        "沪铜": 72000, "沪铝": 19500, "沪锌": 22000, "沪铅": 16000,
        "沪镍": 128000, "沪锡": 250000, "不锈钢": 13500, "国际铜": 65000,
        "氧化铝": 3800, "铸造铝合金期货": 20000, "工业硅": 10500, "碳酸锂": 76000,
        "多晶硅": 42000,
        "螺纹钢": 3500, "热轧卷板": 3600, "线材": 3700, "铁矿石": 850,
        "焦煤": 1350, "焦炭": 2100, "锰硅": 6500, "硅铁": 7200, "动力煤": 850,
        "原油": 550, "燃油": 3200, "沥青": 3600, "橡胶": 14000, "20号胶": 11000,
        "丁二烯橡胶": 12000, "纸浆": 5200, "胶版印刷纸期货": 5500, "低硫燃料油": 3500,
        "PTA": 5800, "郑醇": 2500, "乙二醇": 4500, "尿素": 1700, "纯碱": 1500,
        "玻璃": 1400, "塑料": 8000, "PP": 7500, "PVC": 5600, "苯乙烯": 8200,
        "短纤": 7000, "液化石油气": 4200, "纯苯": 6800, "烧碱": 3200,
        "二甲苯": 8500, "瓶级聚酯切片": 6200, "丙烯": 7500,
        "豆一": 4200, "豆二": 3600, "豆粕": 3000, "豆油": 7800, "棕榈": 7200,
        "菜油": 8500, "菜籽": 5200, "菜粕": 2300, "花生": 8500,
        "棉花": 14000, "棉纱": 22000, "白糖": 6500, "鲜苹果": 8200, "红枣": 10000,
        "玉米": 2200, "玉米淀粉": 2500, "鸡蛋": 3500, "生猪": 14000,
        "粳米": 3200, "粳稻": 2700, "强麦": 2600, "早籼稻": 2500, "晚籼稻": 2600,
        "纤维板": 1200, "胶合板": 160, "原木": 1800,
        "集运指数(欧线)期货": 1200,
    }

    def _init_simulated(self):
        for c in CONTRACTS:
            base = self._BASE.get(c.symbol, 5000.0)
            self._generate_contract(c, base)

    def _generate_contract(self, contract: ContractInfo, base: float):
        sym = contract.symbol
        now = _now()

        for interval in self.INTERVALS:
            count = _BAR_COUNT[interval]
            vol = _VOLATILITY[interval]
            mins = _INTERVAL_MINUTES[interval]
            price = base * (0.9 + np.random.random() * 0.1)
            bars: list[KlineData] = []

            for i in range(count):
                ts = now - timedelta(minutes=mins * (count - i))
                change = np.random.normal(0, vol)
                op = round(price, 2)
                cl = round(price * (1 + change), 2)
                hi = round(max(op, cl) * (1 + abs(np.random.normal(0, vol * 0.4))), 2)
                lo = round(min(op, cl) * (1 - abs(np.random.normal(0, vol * 0.4))), 2)
                vl = int(np.random.exponential(300 * mins) + 50)
                bars.append(KlineData(symbol=sym, interval=interval, open=op, high=hi, low=lo, close=cl, volume=vl, timestamp=ts))
                price = cl

            self._klines[sym][interval] = bars

        final_price = self._klines[sym]["1m"][-1].close
        self._sim_prices[sym] = final_price

        day_bars = self._klines[sym]["1d"]
        recent = day_bars[-1] if day_bars else None
        md = MarketData(
            symbol=sym, name=contract.name_cn, exchange=contract.exchange, category=contract.category,
            open=recent.open if recent else base,
            high=recent.high if recent else base,
            low=recent.low if recent else base,
            close=final_price,
            volume=recent.volume if recent else 0,
            bid=round(final_price - final_price * 0.00005, 2),
            ask=round(final_price + final_price * 0.00005, 2),
            pre_close=round(base, 2),
            change_pct=round((final_price - base) / base, 6),
            timestamp=_now(),
        )
        self._quotes[sym] = md

    async def _tick_sim(self):
        vol = 0.0003
        for c in CONTRACTS:
            sym = c.symbol
            price = self._sim_prices.get(sym, 5000.0)
            change = np.random.normal(0, vol)
            np_ = round(price * (1 + change), 2)
            self._sim_prices[sym] = np_

            hi = round(max(price, np_) * (1 + abs(random.gauss(0, vol * 0.2))), 2)
            lo = round(min(price, np_) * (1 - abs(random.gauss(0, vol * 0.2))), 2)
            vl = int(np.random.exponential(200) + 30)

            kline = KlineData(symbol=sym, interval="1m", open=price, high=hi, low=lo, close=np_, volume=vl, timestamp=_now())
            ks = self._klines[sym]["1m"]
            ks.append(kline)
            if len(ks) > 2000:
                self._klines[sym]["1m"] = ks[-1500:]

            prev = self._quotes.get(sym)
            md = MarketData(
                symbol=sym, name=c.name_cn, exchange=c.exchange, category=c.category,
                open=prev.open if prev else price,
                high=round(max(hi, prev.high if prev else hi), 2),
                low=round(min(lo, prev.low if prev else lo), 2),
                close=np_,
                volume=(prev.volume if prev else 0) + vl,
                bid=round(np_ - np_ * 0.00005, 2), ask=round(np_ + np_ * 0.00005, 2),
                pre_close=round(prev.pre_close if prev else price, 2),
                change_pct=round((np_ - (prev.pre_close if prev else price)) / (prev.pre_close if prev else price), 6),
                timestamp=_now(),
            )
            self._quotes[sym] = md
            await self._notify(md)


market_data_service = MarketDataService()