PorterandCursor
2c1232555e
将核心结构、指标与分析拆到 chan/{core,indicators,analysis,pipeline}；
根目录保留兼容 shim；strategies 改为从 chan 包导入；买卖点经 bsp_macd 与 MACD 接合。
Co-authored-by: Cursor <cursoragent@cursor.com>
42 lines
1.0 KiB
Python
42 lines
1.0 KiB
Python
"""按 bar idx 查询指标值。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, Optional
|
|
|
|
import pandas as pd
|
|
|
|
|
|
class IndicatorStore:
|
|
"""以 DataFrame 列 + 行 idx 对齐的只读指标视图。"""
|
|
|
|
def __init__(self, df: pd.DataFrame):
|
|
self._df = df
|
|
|
|
@property
|
|
def dataframe(self) -> pd.DataFrame:
|
|
return self._df
|
|
|
|
def __len__(self) -> int:
|
|
return len(self._df)
|
|
|
|
def get(self, idx: int, name: str, default: Any = None) -> Any:
|
|
if idx < 0 or idx >= len(self._df):
|
|
return default
|
|
if name not in self._df.columns:
|
|
return default
|
|
val = self._df.iloc[idx][name]
|
|
if pd.isna(val):
|
|
return default
|
|
return val
|
|
|
|
def row(self, idx: int) -> Optional[Dict[str, Any]]:
|
|
if idx < 0 or idx >= len(self._df):
|
|
return None
|
|
return self._df.iloc[idx].to_dict()
|
|
|
|
def series(self, name: str):
|
|
if name not in self._df.columns:
|
|
return None
|
|
return self._df[name]
|