A股实时数据API（记录踩过的坑）

def classify_a_share(symbol: str) -> dict:
    """
    从股票代码识别板块信息，无需调用接口。
    symbol 格式：600519.SH / 300750.SZ / 688599.SH
    """
    code = symbol.split(".")[0]
    suffix = symbol.split(".")[-1] if "." in symbol else ""

    if code.startswith("6") and not code.startswith("688") and not code.startswith("689"):
        board, limit = "沪市主板", 0.10
    elif code.startswith("688") or code.startswith("689"):
        board, limit = "科创板", 0.20
    elif code.startswith("000") or code.startswith("001") or code.startswith("002") or code.startswith("003"):
        board, limit = "深市主板", 0.10
    elif code.startswith("300") or code.startswith("301"):
        board, limit = "创业板", 0.20
    elif code.startswith("83") or code.startswith("87") or code.startswith("43"):
        board, limit = "北交所", 0.30
    else:
        board, limit = "未知", 0.10

    return {"code": code, "suffix": suffix, "board": board, "daily_limit": limit}

print(classify_a_share("600519.SH"))   # 茅台 → 沪市主板, ±10%
print(classify_a_share("300750.SZ"))   # 宁德时代 → 创业板, ±20%
print(classify_a_share("688599.SH"))   # 天合光能 → 科创板, ±20%

import requests

API_KEY = "YOUR_API_KEY"
BASE    = "https://data.infoway.io"

# A 股代码格式：600519.SH（沪市）或 000858.SZ（深市）
codes = "600519.SH,000858.SZ,300750.SZ,688599.SH"

resp = requests.get(
    f"{BASE}/stock/batch_trade/{codes}",
    headers={"apiKey": API_KEY, "Accept": "application/json"}
)
for tick in resp.json()["data"]:
    direction = {0: "—", 1: "↑买", 2: "↓卖"}.get(tick["td"], "?")
    print(f"{tick['s']:<15} 价格={tick['p']:>8}  成交量={tick['v']:>10}手  方向={direction}")

shares = int(tick["v"])
lots   = shares // 100
print(f"成交 {lots} 手 ({shares} 股)")

resp = requests.get(
    f"{BASE}/stock/batch_depth/600519.SH",
    headers={"apiKey": API_KEY, "Accept": "application/json"}
)
book = resp.json()["data"][0]

asks = list(zip(book["a"][0], book["a"][1]))
bids = list(zip(book["b"][0], book["b"][1]))

print("卖盘（ask）")
for price, vol in asks[:5]:
    print(f"  {price:>8} 元   {int(vol)//100:>6} 手")

spread = float(asks[0][0]) - float(bids[0][0])
print(f"\n  价差：{spread:.2f} 元\n")

print("买盘（bid）")
for price, vol in bids[:5]:
    print(f"  {price:>8} 元   {int(vol)//100:>6} 手")

from datetime import time

def is_valid_cn_bar(ts_seconds: int) -> bool:
    """判断秒时间戳（UTC）是否落在 A 股交易时段内（转为北京时间 UTC+8）。"""
    from datetime import datetime, timezone, timedelta
    CST = timezone(timedelta(hours=8))
    t = datetime.fromtimestamp(ts_seconds, tz=CST).time()
    morning   = time(9, 30) <= t < time(11, 30)
    afternoon = time(13, 0) <= t < time(15, 0)
    return morning or afternoon

import requests

def get_adjustment_factors(symbol: str, begin_day: str, end_day: str) -> list[dict]:
    resp = requests.get(
        "https://data.infoway.io/common/basic/symbols/adjustment_factors",
        params={"symbol": symbol, "market": "CN",
                "beginDay": begin_day, "endDay": end_day},
        headers={"apiKey": "YOUR_API_KEY"}
    )
    return resp.json().get("data", [])

factors = get_adjustment_factors("600519.SH", "20200101", "20261231")
# 返回每个交易日的 forward_factor（前复权因子）

# 应用复权因子到 K 线数据
factor_map = {f["trade_date"]: f["forward_factor"] for f in factors}

def apply_forward_adj(candle: dict, trade_date: str) -> dict:
    factor = factor_map.get(trade_date, 1.0)
    return {
        **candle,
        "o": str(round(float(candle["o"]) * factor, 4)),
        "h": str(round(float(candle["h"]) * factor, 4)),
        "l": str(round(float(candle["l"]) * factor, 4)),
        "c": str(round(float(candle["c"]) * factor, 4)),
    }

import requests, json, time as _time

def fetch_cn_klines(symbol: str, kline_type: int = 8,
                    count: int = 500, until_ts: int | None = None) -> list[dict]:
    payload = {"klineType": kline_type, "klineNum": count, "codes": symbol}
    if until_ts:
        payload["timestamp"] = until_ts
    r = requests.post(
        "https://data.infoway.io/stock/v2/batch_kline",
        headers={"apiKey": "YOUR_API_KEY", "Content-Type": "application/json"},
        data=json.dumps(payload)
    )
    r.raise_for_status()
    for item in r.json().get("data", []):
        if item["s"] == symbol:
            return item.get("respList", [])
    return []

def download_full_history(symbol: str) -> list[dict]:
    """翻页下载全部日 K 线历史。"""
    all_bars: list[dict] = []
    cursor: int | None = None

    while True:
        batch = fetch_cn_klines(symbol, kline_type=8, count=500, until_ts=cursor)
        if not batch:
            break
        all_bars.extend(batch)
        oldest = int(batch[-1]["t"])
        if cursor and oldest >= cursor:
            break
        cursor = oldest
        _time.sleep(0.3)

    all_bars.sort(key=lambda b: int(b["t"]))
    return all_bars

bars = download_full_history("600519.SH")
print(f"茅台日K共 {len(bars)} 根，最早 {bars[0]['t']}，最新 {bars[-1]['c']} 元")

import math

def calc_limit_prices(prev_close: float, board: str,
                      is_st: bool = False) -> tuple[float, float]:
    """
    返回 (涨停价, 跌停价)。
    board 参考接口返回的 board 字段：
      SHMainConnect / SHMainNonConnect / SZMainConnect / SZMainNonConnect
      SHSTAR / SZGEMConnect / SZGEMNonConnect
    """
    if is_st:
        pct = 0.05
    elif board in ("SHSTAR", "SZGEMConnect", "SZGEMNonConnect"):
        pct = 0.20
    else:
        pct = 0.10

    # 交易所规则：涨停价向下取整到分，跌停价向上取整到分
    up   = math.floor(prev_close * (1 + pct) * 100) / 100
    down = math.ceil(prev_close  * (1 - pct) * 100) / 100
    return up, down

# 示例：茅台前收 1800.00 元，主板 ±10%
up, down = calc_limit_prices(1800.00, "SHMainConnect")
print(f"涨停价：{up:.2f}  跌停价：{down:.2f}")
# → 涨停价：1980.00  跌停价：1620.00

def is_limit(current_price: float, prev_close: float,
             board: str, is_st: bool = False) -> str:
    up, down = calc_limit_prices(prev_close, board, is_st)
    if abs(current_price - up) < 0.005:
        return "涨停"
    if abs(current_price - down) < 0.005:
        return "跌停"
    return "正常"

wss://data.infoway.io/ws?business=stock&apikey=YOUR_API_KEY

import asyncio, json, uuid, logging
from datetime import datetime, time, timezone, timedelta
from typing import Optional
import websockets
from websockets.exceptions import ConnectionClosed

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("a-share-ws")

CST = timezone(timedelta(hours=8))

def cn_market_status() -> str:
    """返回当前 A 股市场状态。"""
    now = datetime.now(CST)
    if now.weekday() >= 5:
        return "休市"
    t = now.time()
    if time(9, 15) <= t < time(9, 30):
        return "集合竞价"
    if time(9, 30) <= t < time(11, 30):
        return "上午交易"
    if time(11, 30) <= t < time(13, 0):
        return "午休"
    if time(13, 0) <= t < time(14, 57):
        return "下午交易"
    if time(14, 57) <= t < time(15, 0):
        return "收盘竞价"
    return "休市"

# 监控的股票列表（示例：茅台、宁德时代、中芯国际、比亚迪）
SYMBOLS = "600519.SH,300750.SZ,688981.SH,002594.SZ"


class AShareClient:
    def __init__(self, api_key: str):
        self.url = f"wss://data.infoway.io/ws?business=stock&apikey={api_key}"
        self.ws: Optional[websockets.WebSocketClientProtocol] = None
        self.running = True
        self._hb_task: Optional[asyncio.Task] = None

    def _trace(self) -> str:
        return str(uuid.uuid4())

    async def _send(self, msg: dict) -> None:
        if self.ws:
            await self.ws.send(json.dumps(msg))

    async def _subscribe(self) -> None:
        # 订阅实时成交（代码 10000）
        await self._send({"code": 10000, "trace": self._trace(),
                          "data": {"codes": SYMBOLS}})
        await asyncio.sleep(5)

        # 订阅盘口（代码 10003）
        await self._send({"code": 10003, "trace": self._trace(),
                          "data": {"codes": SYMBOLS}})
        await asyncio.sleep(5)

        # 订阅 1 分钟 K 线（代码 10006）
        await self._send({"code": 10006, "trace": self._trace(),
                          "data": {"arr": [{"type": 1, "codes": SYMBOLS}]}})

        log.info("订阅完成 | 市场状态：%s", cn_market_status())

    def _start_heartbeat(self) -> None:
        if self._hb_task and not self._hb_task.done():
            self._hb_task.cancel()

        async def beat():
            while True:
                await asyncio.sleep(30)
                if not self.ws or self.ws.close_code is not None:
                    break
                try:
                    await self._send({"code": 10010, "trace": self._trace()})
                except Exception:
                    break

        self._hb_task = asyncio.create_task(beat())

    def _dispatch(self, raw: str) -> None:
        try:
            msg = json.loads(raw)
        except json.JSONDecodeError:
            return

        code = msg.get("code")
        data = msg.get("data", {})

        if code == 10002:   # 成交推送
            direction = {0: "—", 1: "↑主买", 2: "↓主卖"}.get(data.get("td"), "?")
            vol_lots  = int(data.get("v", 0)) // 100
            log.info("成交 %-14s 价格=%-9s 成交量=%-6d手 %s",
                     data.get("s"), data.get("p"), vol_lots, direction)

        elif code == 10005: # 盘口推送
            best_bid = data.get("b", [[None]])[0][0]
            best_ask = data.get("a", [[None]])[0][0]
            if best_bid and best_ask:
                spread = round(float(best_ask) - float(best_bid), 2)
                log.info("盘口 %-14s 买一=%-9s 卖一=%-9s 价差=%.2f",
                         data.get("s"), best_bid, best_ask, spread)

        elif code == 10008: # K 线推送
            pct = data.get("pfr", "0%")
            log.info("K线  %-14s 开=%-7s 高=%-7s 低=%-7s 收=%-7s 涨跌=%s",
                     data.get("s"), data.get("o"), data.get("h"),
                     data.get("l"), data.get("c"), pct)

        elif code in (10001, 10004, 10007):
            log.info("订阅确认 code=%s", code)

    async def _connect_once(self) -> None:
        async with websockets.connect(self.url) as ws:
            self.ws = ws
            log.info("已连接 | 市场状态：%s", cn_market_status())
            await self._subscribe()
            self._start_heartbeat()
            try:
                async for message in ws:
                    self._dispatch(message)
            finally:
                if self._hb_task:
                    self._hb_task.cancel()
                self.ws = None

    async def start(self) -> None:
        backoff = 5
        while self.running:
            status = cn_market_status()
            # 午休和休市期间暂停重连，避免无意义的断线重连
            if status in ("午休", "休市"):
                log.info("当前状态：%s，暂停连接（60s后再检查）", status)
                await asyncio.sleep(60)
                continue
            try:
                await self._connect_once()
                backoff = 5
            except ConnectionClosed as e:
                log.warning("连接断开：%s", e)
            except Exception as e:
                log.error("连接异常：%s", e)
            if not self.running:
                break
            log.info("%ds 后重连...", backoff)
            await asyncio.sleep(backoff)
            backoff = min(backoff * 2, 60)

    def stop(self) -> None:
        self.running = False


async def main():
    client = AShareClient(api_key="YOUR_API_KEY")
    await client.start()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        log.info("已停止")

from collections import defaultdict
from datetime import date

class T1PositionTracker:
    """跟踪 A 股 T+1 约束：今天买入的不能今天卖出。"""

    def __init__(self):
        self.positions: dict[str, int] = defaultdict(int)       # symbol → 可卖数量（股）
        self.today_bought: dict[str, int] = defaultdict(int)    # 今日新买，次日才能卖

    def buy(self, symbol: str, shares: int) -> None:
        self.today_bought[symbol] += shares

    def sell(self, symbol: str, shares: int) -> bool:
        available = self.positions[symbol]
        if shares > available:
            print(f"⚠ {symbol} 可卖 {available} 股，不足 {shares} 股（T+1 限制）")
            return False
        self.positions[symbol] -= shares
        return True

    def next_day(self) -> None:
        """每日收盘后调用，将今日买入转为可卖持仓。"""
        for symbol, shares in self.today_bought.items():
            self.positions[symbol] += shares
        self.today_bought.clear()