Upload 4 files

插件_行情/K线聚合器.py

@@ -0,0 +1,72 @@
+# -*- coding: utf-8 -*-
+from typing import Optional, List, Dict, Iterable
+from .类型定义 import 蜡烛, 成交
+
+def _分钟起点(ms: int) -> int:
+    return (ms // 60000) * 60000
+
+class 一分钟K线生成器:
+    """用成交流生成1mK；每分钟切换时封口上一根"""
+    def __init__(self, 合约: str):
+        self.合约 = 合约
+        self._分钟: Optional[int] = None
+        self._开 = self._高 = self._低 = self._收 = None
+        self._量 = 0.0
+
+    def 更新成交(self, t: 成交) -> Optional[蜡烛]:
+        m = _分钟起点(t.时间戳)
+        if self._分钟 is not None and m > self._分钟:
+            完 = 蜡烛(self.合约, '1m', self._分钟, float(self._开), float(self._高), float(self._低), float(self._收), float(self._量), True, {})
+            self._分钟 = m
+            self._开 = self._高 = self._低 = self._收 = t.价
+            self._量 = abs(t.量)
+            return 完
+        if self._分钟 is None:
+            self._分钟 = m
+            self._开 = self._高 = self._低 = self._收 = t.价
+            self._量 = abs(t.量)
+        else:
+            self._收 = t.价
+            self._高 = max(self._高, t.价)
+            self._低 = min(self._低, t.价)
+            self._量 += abs(t.量)
+        return None
+
+    def 当前未封口(self) -> Optional[蜡烛]:
+        if self._分钟 is None: return None
+        return 蜡烛(self.合约,'1m',self._分钟,float(self._开),float(self._高),float(self._低),float(self._收),float(self._量),False,{})
+
+class 多周期聚合器:
+    """把已收盘的1mK聚合到更大周期；仅在1m收盘时推进"""
+    def __init__(self, 合约: str, 周期集合_分钟: Iterable[int]):
+        self.合约 = 合约
+        self.周期 = sorted(set(int(x) for x in 周期集合_分钟))
+        self._缓冲: Dict[int, Dict[str, float]] = {}  # tf: {'开始':ts,'开':,'高':,'低':,'收':,'量':}
+
+    def _周期起点(self, ts_ms: int, tf: int) -> int:
+        分 = ts_ms // 60000
+        起 = (分 // tf) * tf
+        return 起 * 60000
+
+    def 接收_1m收盘(self, 一分K: 蜡烛) -> List[蜡烛]:
+        assert 一分K.周期 == '1m' and 一分K.已收盘
+        输出: List[蜡烛] = []
+        for tf in self.周期:
+            起 = self._周期起点(一分K.时间戳, tf)
+            buf = self._缓冲.get(tf)
+            if buf is None or buf['开始'] != 起:
+                if buf is not None:
+                    输出.append(蜡烛(self.合约, f'{tf}m', int(buf['开始']), float(buf['开']), float(buf['高']), float(buf['低']), float(buf['收']), float(buf['量']), True, {}))
+                self._缓冲[tf] = {'开始': 起, '开': 一分K.开, '高': 一分K.高, '低': 一分K.低, '收': 一分K.收, '量': 一分K.量}
+            else:
+                buf['收'] = 一分K.收
+                buf['高'] = max(buf['高'], 一分K.高)
+                buf['低'] = min(buf['低'], 一分K.低)
+                buf['量'] += 一分K.量
+        return 输出
+
+    def 当前未封口(self) -> List[蜡烛]:
+        res=[]
+        for tf, b in self._缓冲.items():
+            res.append(蜡烛(self.合约, f'{tf}m', int(b['开始']), float(b['开']), float(b['高']), float(b['低']), float(b['收']), float(b['量']), False, {}))
+        return res

插件_行情/公共WS.py

@@ -0,0 +1,91 @@
+# -*- coding: utf-8 -*-
+import asyncio, json, time
+from typing import List, Dict, Any, Callable, Awaitable, Optional
+import websockets
+
+OKX公共WS地址 = "wss://ws.okx.com:8443/ws/v5/public"
+
+class OKX公共WS:
+    """
+    - add: 添加订阅(channel='candle1m'/'trades'/'tickers', instId='BTC-USDT-SWAP')
+    - on:  注册回调(统一事件字典)
+    - run: 永久运行(自动重连/心跳/续订)
+    事件格式:
+      {'类型': 'k线'|'成交'|'系统'|'确认'|'错误',
+       '频道': 'candle1m'|'trades'|'tickers',
+       '合约': 'BTC-USDT-SWAP',
+       '参数': {...}, '数据': [...], '时间': int?}
+    """
+    def __init__(self, 地址: str = OKX公共WS地址, 心跳秒: int = 20, 重连秒: int = 3):
+        self.地址 = 地址
+        self.心跳秒 = 心跳秒
+        self.重连秒 = 重连秒
+        self._订阅: List[Dict[str, Any]] = []
+        self._回调: List[Callable[[Dict[str, Any]], Awaitable[None] | None]] = []
+        self._ws = None
+        self._停 = False
+
+    def 添加订阅(self, 频道: str, 合约: str, bar: Optional[str] = None):
+        if 频道.startswith('candle') and bar and not 频道.endswith(bar):
+            频道 = f'candle{bar}'
+        self._订阅.append({"channel": 频道, "instId": 合约})
+
+    def 注册回调(self, 函数: Callable[[Dict[str, Any]], Awaitable[None] | None]):
+        self._回调.append(函数)
+
+    async def _发(self, ws, msg: Dict[str, Any]):
+        await ws.send(json.dumps(msg))
+
+    async def _发心跳(self, ws):
+        while not self._停:
+            try:
+                await ws.ping()
+            except Exception:
+                return
+            await asyncio.sleep(self.心跳秒)
+
+    async def _续订(self, ws):
+        if not self._订阅:
+            return
+        批 = 20
+        for i in range(0, len(self._订阅), 批):
+            args = self._订阅[i:i+批]
+            await self._发(ws, {"op": "subscribe", "args": args})
+            await asyncio.sleep(0.05)
+
+    async def _emit(self, 事件: Dict[str, Any]):
+        for cb in list(self._回调):
+            r = cb(事件)
+            if asyncio.iscoroutine(r):
+                await r
+
+    async def 永久运行(self):
+        while not self._停:
+            try:
+                async with websockets.connect(self.地址, ping_interval=None, close_timeout=3) as ws:
+                    self._ws = ws
+                    await self._emit({'类型':'系统','事件':'已连接'})
+                    await self._续订(ws)
+                    心跳 = asyncio.create_task(self._发心跳(ws))
+                    try:
+                        async for 原 in ws:
+                            try:
+                                m = json.loads(原)
+                            except Exception:
+                                continue
+                            if isinstance(m, dict) and 'event' in m:
+                                ev = '确认' if m['event']=='subscribe' else '错误'
+                                await self._emit({'类型':ev,'参数':m.get('arg'),'代码':m.get('code'),'信息':m.get('msg')})
+                                continue
+                            if 'arg' in m and 'data' in m:
+                                arg = m['arg']; ch = arg.get('channel'); inst = arg.get('instId')
+                                类 = 'k线' if str(ch).startswith('candle') else ('成交' if ch=='trades' else '数据')
+                                await self._emit({'类型':类,'频道':ch,'合约':inst,'参数':arg,'数据':m['data']})
+                    finally:
+                        心跳.cancel()
+            except Exception as e:
+                await self._emit({'类型':'系统','事件':'断开','错误':str(e)})
+                await asyncio.sleep(self.重连秒)
+
+    def 停止(self):
+        self._停 = True

插件_行情/历史数据.py

@@ -0,0 +1,40 @@
+# -*- coding: utf-8 -*-
+from typing import List, Optional
+import aiohttp, pandas as pd
+from .类型定义 import 蜡烛
+
+OKX_REST根 = "https://www.okx.com"
+
+async def 获取OKX历史1分钟K(合约: str, 根数: int = 1000) -> List[蜡烛]:
+    url = f"{OKX_REST根}/api/v5/market/history-candles"
+    params = {"instId": 合约, "bar": "1m", "limit": str(根数)}
+    out: List[蜡烛] = []
+    async with aiohttp.ClientSession() as sess:
+        async with sess.get(url, params=params, timeout=20) as r:
+            r.raise_for_status()
+            js = await r.json()
+            for row in js.get('data', []):
+                ts = int(row[0]); o,h,l,c,vol = map(float, (row[1],row[2],row[3],row[4],row[5]))
+                out.append(蜡烛(合约,'1m',ts,o,h,l,c,vol,True,{}))
+    out.sort(key=lambda x: x.时间戳)
+    return out
+
+def 读取CSV为1分钟K(路径: str, 合约: str,
+                 时间列='时间', 开列='开', 高列='高', 低列='低', 收列='收', 量列='量') -> List[蜡烛]:
+    df = pd.read_csv(路径)
+    if 时间列 in df.columns:
+        ts = pd.to_datetime(df[时间列], utc=True, errors='coerce').astype('int64')//10**6
+        df = df.assign(_ts=ts)
+    else:
+        for c in ['timestamp','Datetime','date']:
+            if c in df.columns:
+                ts = pd.to_datetime(df[c], utc=True, errors='coerce').astype('int64')//10**6
+                df = df.assign(_ts=ts); break
+    需要 = [开列,高列,低列,收列]
+    for c in 需要:
+        if c not in df.columns:
+            raise ValueError(f"CSV 缺少列: {c}")
+    out=[]
+    for _,r in df.sort_values('_ts').iterrows():
+        out.append(蜡烛(合约,'1m',int(r['_ts']), float(r[开列]), float(r[高列]), float(r[低列]), float(r[收列]), float(r.get(量列,0.0)), True, {}))
+    return out

插件_行情/类型定义.py

@@ -0,0 +1,30 @@
+# -*- coding: utf-8 -*-
+from __future__ import annotations
+from dataclasses import dataclass, asdict
+from typing import Optional, Dict, Any
+
+@dataclass
+class 蜡烛:
+    合约: str          # 例: "BTC-USDT-SWAP"
+    周期: str          # '1m','60m','120m','240m'
+    时间戳: int        # 毫秒
+    开: float
+    高: float
+    低: float
+    收: float
+    量: float
+    已收盘: bool
+    附加: Optional[Dict[str, Any]] = None
+
+    def 到字典(self) -> Dict[str, Any]:
+        d = asdict(self)
+        d['time'] = d.pop('时间戳')
+        return d
+
+@dataclass
+class 成交:
+    合约: str
+    时间戳: int
+    价: float
+    量: float
+    方向: str  # 'buy'/'sell'（可能缺省）