模拟盘无 API Key 时复盘K线改为优先走 OKX 公开行情接口。
Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,40 @@
|
||||
"""OKX 公开 K 线小测(无网络):instId / bar 映射."""
|
||||
from __future__ import annotations
|
||||
|
||||
import unittest
|
||||
|
||||
|
||||
class OkxPublicOhlcvMappingTests(unittest.TestCase):
|
||||
def test_swap_inst_and_bar(self):
|
||||
# 避免 import 整站 app:直接复刻映射逻辑做契约校验
|
||||
def swap_inst(exchange_symbol: str) -> str:
|
||||
sym = str(exchange_symbol or "").strip().upper()
|
||||
if not sym:
|
||||
return ""
|
||||
if "/" in sym:
|
||||
base, rest = sym.split("/", 1)
|
||||
quote = rest.split(":", 1)[0].strip() or "USDT"
|
||||
return f"{base.strip()}-{quote}-SWAP"
|
||||
if sym.endswith("-SWAP"):
|
||||
return sym
|
||||
return sym
|
||||
|
||||
def public_bar(timeframe: str) -> str:
|
||||
low = str(timeframe or "").strip().lower()
|
||||
if low.endswith("m") and low[:-1].isdigit():
|
||||
return low
|
||||
if low.endswith("h") and low[:-1].isdigit():
|
||||
return f"{low[:-1]}H"
|
||||
if low.endswith("d") and low[:-1].isdigit():
|
||||
return f"{low[:-1]}D"
|
||||
return low or "15m"
|
||||
|
||||
self.assertEqual(swap_inst("ETH/USDT:USDT"), "ETH-USDT-SWAP")
|
||||
self.assertEqual(swap_inst("BTC/USDT:USDT"), "BTC-USDT-SWAP")
|
||||
self.assertEqual(public_bar("15m"), "15m")
|
||||
self.assertEqual(public_bar("1h"), "1H")
|
||||
self.assertEqual(public_bar("4h"), "4H")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user