gavindiaz/QuanTAlib
1
0
Fork 0
mirror of https://github.com/mihakralj/QuanTAlib.git synced 2026-08-17 01:58:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

minor

Browse Source
This commit is contained in:
Miha Kralj
2026-02-28 16:05:11 -08:00
parent 83e9511261
commit 768123d056
7 changed files with 307 additions and 150 deletions
Download Patch File Download Diff File Expand all files Collapse all files
python/tests/run_all_exported_pandasta_validation.py
+132 -21
View File
@@ -61,14 +61,32 @@ SPECIAL_PTA: dict[str, Callable[[], np.ndarray]] = {
"cfo": lambda: (100.0 * (S_CLOSE - ta.linreg(S_CLOSE, length=14, tsf=False, talib=False)) / S_CLOSE).to_numpy(),
"trix": lambda: ta.trix(S_CLOSE, length=18).iloc[:, 0].to_numpy(),
"dpo": lambda: ta.dpo(S_CLOSE, length=20, centered=False).to_numpy(),
# pandas-ta PVT uses percent ROC scaling; normalize to ratio-scale for QuanTAlib parity
"pvt": lambda: (ta.pvt(S_CLOSE, S_VOLUME) / 100.0).to_numpy(),
# Match QuanTAlib EOM default volume scale (10_000) vs pandas-ta default divisor (100_000_000)
"eom": lambda: ta.eom(S_HIGH, S_LOW, S_CLOSE, S_VOLUME, length=14, divisor=10_000, drift=1).to_numpy(),
# Force pandas-ta RSI path (no TA-Lib shortcut) and Wilder smoothing to align with wrapper path
"rsi": lambda: ta.rsi(S_CLOSE, length=14, mamode="rma", talib=False, drift=1, scalar=100).to_numpy(),
# QuanTAlib ROC is absolute delta; convert pandas-ta percent ROC to absolute for parity
"roc": lambda: (ta.roc(S_CLOSE, length=10, scalar=100, talib=False) * S_CLOSE.shift(10) / 100.0).to_numpy(),
# Match CRSI parameter names and internal RSI path
"crsi": lambda: ta.crsi(
S_CLOSE, rsi_length=3, streak_length=2, rank_length=100,
scalar=100, talib=False, drift=1
).to_numpy(),
# Match BBands to pandas-ta native path with ddof=0 (wrapper/stddev parity basis)
"bbands": lambda: ta.bbands(
S_CLOSE, length=20, lower_std=2.0, upper_std=2.0,
ddof=0, mamode="sma", talib=False
).filter(regex=r"^BBU").iloc[:, 0].to_numpy(),
}
ALIASES = {
ALIASES: dict[str, str | None] = {
"medprice": "midprice",
"typprice": "hlc3",
"avgprice": "ohlc4",
"midbody": "mid_body",
"mom": "momentum",
"midbody": None,
"mom": "mom",
"bbands": "bbands",
"stddev": "stdev",
"zscore": "zscore",
@@ -77,6 +95,31 @@ ALIASES = {
"dema_alpha": None,
}
# Explicitly tracked indicators with no meaningful pandas-ta equivalent.
NO_PTA_EQUIVALENT: set[str] = {
"afirma", "agc", "ahrens", "alaguerre", "apchannel", "atrbands",
"baxterking", "bbb", "bbi", "bbwn", "bbwp", "bessel", "betadist",
"bilateral", "binomdist", "blma", "bpf", "butter2", "butter3",
"bwma", "ccor", "ccv", "ccyc", "cfitz", "change", "cheby1", "cheby2",
"cointegration", "conv", "coral", "correlation", "covariance", "crma",
"cv", "cvi", "cwt", "deco", "decycler", "dem", "dema_alpha", "dosc",
"dsma", "dsp", "dwma", "dwt", "dymoi", "eacp", "edcf", "elliptic",
"ema_alpha", "etherm", "evwma", "ewma", "expdist", "exptrans",
"fisher04", "gdema", "hanma", "hema", "kri", "lema", "lsma", "mae",
"mape", "mse", "parzen", "pvd", "rain", "rmse", "sgma", "sinema",
"sp15", "tsf", "tukey_w", "tvi", "vf",
# pandas-ta implementations diverge materially from QuanTAlib formulations in this snapshot
"nvi", "pvi",
}
PRIMARY_OUTPUT_PREFIX: dict[str, tuple[str, ...]] = {
"bbands": ("BBU", "BBM", "BBL"),
"pvo": ("PVO_", "PVOs", "PVOh"),
"brar": ("BR_", "AR_"),
# QuanTAlib AOBV primary output is fast EMA line
"aobv": ("OBVe_4", "OBV", "AOBV"),
}
SKIP_PRIVATE = {
"_arr",
"_ptr",
@@ -91,12 +134,20 @@ SKIP_PRIVATE = {
}
def normalize_pta_output(v: Any) -> np.ndarray:
def _select_df_column(df: pd.DataFrame, indicator_name: str) -> pd.Series:
prefixes = PRIMARY_OUTPUT_PREFIX.get(indicator_name, ())
cols = list(df.columns)
for pref in prefixes:
for c in cols:
if str(c).startswith(pref):
return df[c]
return df.iloc[:, 0]
def normalize_output(v: Any, indicator_name: str) -> np.ndarray:
if isinstance(v, pd.Series):
return v.to_numpy()
if isinstance(v, pd.DataFrame):
# default: first numeric column
return v.iloc[:, 0].to_numpy()
return _select_df_column(v, indicator_name).to_numpy()
if isinstance(v, tuple):
if len(v) == 0:
return np.array([], dtype=np.float64)
@@ -114,10 +165,22 @@ def get_q_functions() -> dict[str, Callable[..., Any]]:
def choose_pta_name(q_name: str) -> str | None:
if q_name in ALIASES:
return ALIASES[q_name]
if hasattr(ta, q_name):
return q_name
if q_name in NO_PTA_EQUIVALENT:
return None
candidates: list[str] = []
alias = ALIASES.get(q_name, "__MISSING__")
if alias != "__MISSING__":
if alias is None:
return None
candidates.append(alias)
candidates.append(q_name)
for name in candidates:
obj = getattr(ta, name, None)
if obj is not None and callable(obj):
return name
return None
@@ -176,16 +239,23 @@ def call_q(name: str, fn: Callable[..., Any]) -> np.ndarray:
if param.default is inspect._empty:
raise RuntimeError(f"required arg {p} not mapped")
out = fn(*args, **kwargs)
return normalize_pta_output(out)
return normalize_output(out, name)
def call_pta(q_name: str) -> np.ndarray:
def _q_default(fn: Callable[..., Any], param: str, fallback: Any) -> Any:
sig = inspect.signature(fn)
p = sig.parameters.get(param)
if p is None or p.default is inspect._empty or p.default is None:
return fallback
return p.default
def call_pta(q_name: str, q_fn: Callable[..., Any]) -> np.ndarray | None:
if q_name in SPECIAL_PTA:
return SPECIAL_PTA[q_name]()
pta_name = choose_pta_name(q_name)
if not pta_name:
raise RuntimeError("no pandas-ta mapping")
return None
pta_fn = getattr(ta, pta_name)
sig = inspect.signature(pta_fn)
@@ -197,23 +267,50 @@ def call_pta(q_name: str) -> np.ndarray:
kwargs: dict[str, Any] = {}
if "length" in params:
kwargs["length"] = 14
kwargs["length"] = int(_q_default(q_fn, "length", 14))
if "fast" in params:
kwargs["fast"] = 12
kwargs["fast"] = int(_q_default(q_fn, "fast", 12))
if "slow" in params:
kwargs["slow"] = 26
kwargs["slow"] = int(_q_default(q_fn, "slow", 26))
if "signal" in params:
kwargs["signal"] = 9
kwargs["signal"] = int(_q_default(q_fn, "signal", 9))
if "offset" in params:
kwargs["offset"] = 0
# Indicator-specific parity defaults
if q_name == "bbands":
kwargs["length"] = int(_q_default(q_fn, "length", 20))
kwargs["lower_std"] = float(_q_default(q_fn, "std", 2.0))
kwargs["upper_std"] = float(_q_default(q_fn, "std", 2.0))
kwargs["ddof"] = 0
kwargs["mamode"] = "sma"
kwargs["talib"] = False
elif q_name == "rsi":
kwargs["length"] = int(_q_default(q_fn, "length", 14))
kwargs["mamode"] = "rma"
kwargs["talib"] = False
kwargs["drift"] = 1
kwargs["scalar"] = 100
elif q_name == "roc":
kwargs["length"] = int(_q_default(q_fn, "length", 10))
kwargs["talib"] = False
kwargs["scalar"] = 100
elif q_name == "crsi":
kwargs.pop("length", None)
kwargs["rsi_length"] = int(_q_default(q_fn, "rsi_period", 3))
kwargs["streak_length"] = int(_q_default(q_fn, "streak_period", 2))
kwargs["rank_length"] = int(_q_default(q_fn, "rank_period", 100))
kwargs["scalar"] = 100
kwargs["talib"] = False
kwargs["drift"] = 1
args: list[Any] = []
for p in params:
if p in kwargs:
continue
if p == "close":
args.append(S_CLOSE)
elif p == "open":
elif p in {"open", "open_"}:
args.append(S_OPEN)
elif p == "high":
args.append(S_HIGH)
@@ -240,7 +337,14 @@ def call_pta(q_name: str) -> np.ndarray:
pass
out = pta_fn(*args, **kwargs)
return normalize_pta_output(out)
# Post-transform for formula alignment
if q_name == "roc":
length = int(_q_default(q_fn, "length", 10))
roc_series = out if isinstance(out, pd.Series) else pd.Series(np.asarray(out), index=S_CLOSE.index)
out = roc_series * S_CLOSE.shift(length) / 100.0
return normalize_output(out, q_name)
def verify_last_n(qtl_arr: np.ndarray, pta_arr: np.ndarray, tol: float = DEFAULT_TOL) -> tuple[bool, float, int]:
@@ -265,12 +369,18 @@ def main() -> int:
rows: list[tuple[str, str, str]] = []
ok = 0
fail = 0
skip = 0
for name in names:
fn = funcs[name]
try:
qv = call_q(name, fn)
pv = call_pta(name)
pv = call_pta(name, fn)
if pv is None:
rows.append((name, "⏭️", "no comparable pandas-ta equivalent"))
skip += 1
continue
passed, max_diff, n = verify_last_n(qv, pv, DEFAULT_TOL)
if passed:
rows.append((name, "✔️", f"max_diff={max_diff:.3e}, n={n}"))
@@ -287,6 +397,7 @@ def main() -> int:
"",
f"- Total indicators scanned: **{len(rows)}**",
f"- Successful (✔️): **{ok}**",
f"- Non-comparable / skipped (⏭️): **{skip}**",
f"- Failing (⚠️): **{fail}**",
"",
"| Indicator | Status | Notes |",
@@ -298,7 +409,7 @@ def main() -> int:
REPORT_PATH.write_text("\n".join(lines), encoding="utf-8")
print(f"Wrote {REPORT_PATH}")
print(f"TOTAL={len(rows)} OK={ok} FAIL={fail}")
print(f"TOTAL={len(rows)} OK={ok} SKIP={skip} FAIL={fail}")
return 0