dekun
92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
|
|
def field_float(item: list[str], idx: int) -> float | None:
|
|
"""Parse one K-line field; empty or invalid strings return None."""
|
|
if idx >= len(item):
|
|
return None
|
|
raw = str(item[idx]).strip()
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return float(raw)
|
|
except (TypeError, ValueError):
|
|
return None
|
|
|
|
|
|
def rows_to_ohlcv(
|
|
rows: list[list[str]],
|
|
*,
|
|
require_volume: bool = True,
|
|
) -> tuple[list[float], list[float], list[float], list[float], list[float]]:
|
|
"""
|
|
Gate candle rows: [ts_ms, o, h, l, c, v, ...].
|
|
Skips rows with missing/invalid OHLC; skips rows with missing volume when require_volume=True.
|
|
"""
|
|
o, h, l, c, v = [], [], [], [], []
|
|
for item in rows:
|
|
if len(item) < 5:
|
|
continue
|
|
o_val = field_float(item, 1)
|
|
h_val = field_float(item, 2)
|
|
l_val = field_float(item, 3)
|
|
c_val = field_float(item, 4)
|
|
if o_val is None or h_val is None or l_val is None or c_val is None:
|
|
continue
|
|
vol_val = field_float(item, 5) if len(item) > 5 else None
|
|
if require_volume and vol_val is None:
|
|
continue
|
|
o.append(o_val)
|
|
h.append(h_val)
|
|
l.append(l_val)
|
|
c.append(c_val)
|
|
v.append(vol_val if vol_val is not None else 0.0)
|
|
return o, h, l, c, v
|
|
|
|
|
|
def rows_to_hlc(rows: list[list[str]]) -> tuple[list[float], list[float], list[float]]:
|
|
"""High / low / close series aligned to valid OHLC rows."""
|
|
h, l_, c = [], [], []
|
|
for item in rows:
|
|
if len(item) < 5:
|
|
continue
|
|
h_val = field_float(item, 2)
|
|
l_val = field_float(item, 3)
|
|
c_val = field_float(item, 4)
|
|
if h_val is None or l_val is None or c_val is None:
|
|
continue
|
|
h.append(h_val)
|
|
l_.append(l_val)
|
|
c.append(c_val)
|
|
return h, l_, c
|
|
|
|
|
|
def rows_to_ohl(rows: list[list[str]]) -> tuple[list[float], list[float], list[float], list[float]]:
|
|
"""Open / high / low / close for chart rendering."""
|
|
o, h, l, c = [], [], [], []
|
|
for item in rows:
|
|
if len(item) < 5:
|
|
continue
|
|
o_val = field_float(item, 1)
|
|
h_val = field_float(item, 2)
|
|
l_val = field_float(item, 3)
|
|
c_val = field_float(item, 4)
|
|
if o_val is None or h_val is None or l_val is None or c_val is None:
|
|
continue
|
|
o.append(o_val)
|
|
h.append(h_val)
|
|
l.append(l_val)
|
|
c.append(c_val)
|
|
return o, h, l, c
|
|
|
|
|
|
def rows_to_close(rows: list[list[str]]) -> list[float]:
|
|
out: list[float] = []
|
|
for item in rows:
|
|
if len(item) < 5:
|
|
continue
|
|
c_val = field_float(item, 4)
|
|
if c_val is not None:
|
|
out.append(c_val)
|
|
return out
|