test(tools): add --profile mode to bench_indicators

cProfile the full indicator run and rank hot functions by tottime and ncalls.
Call-count ranking surfaces per-call overhead (Python callbacks, rolling.apply,
per-bar loops) that wall-clock timing under-weights - which is how the rolling
argmax/argmin and FIR-dot bottlenecks were identified. Usage:
python tools/bench_indicators.py --profile [N] (default N=25).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
rmarcink
2026-07-13 12:49:55 +02:00
co-authored by Claude Opus 4.8
parent c5d60b21ca
commit 7675f4b4de
+47
View File
@@ -111,12 +111,55 @@ def bench(names: list[str], rows: int, repeats: int) -> list[dict]:
return results
def profile(names: list[str], rows: int, top: int) -> None:
"""cProfile the whole indicator set once and rank hot functions.
Surfaces per-call overhead (Python callbacks, rolling.apply, per-bar loops)
that a pure wall-clock ranking under-weights. High ``ncalls`` on a project
function is a vectorise/njit candidate.
"""
import cProfile
import pstats
df = make_ohlcv(rows)
accs = [getattr(df.ta, n, None) for n in names]
accs = [a for a in accs if a is not None]
for a in accs: # warm up (numba JIT + import side effects out of the profile)
try:
a()
except Exception: # noqa: BLE001
pass
pr = cProfile.Profile()
pr.enable()
for a in accs:
try:
a()
except Exception: # noqa: BLE001
pass
pr.disable()
stats = pstats.Stats(pr)
for sort_key in ("tottime", "ncalls"):
print(f"\n=== top {top} by {sort_key} ===")
stats.sort_stats(sort_key).print_stats(top)
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("names", nargs="*", help="indicator names to benchmark (default: all)")
parser.add_argument("--rows", type=int, default=5000, help="number of bars (default: 5000)")
parser.add_argument("--repeats", type=int, default=7, help="timed runs per indicator (default: 7)")
parser.add_argument("--top", type=int, default=None, help="print only the N slowest (default: all)")
parser.add_argument(
"--profile",
type=int,
nargs="?",
const=25,
default=None,
metavar="N",
help="cProfile the run and print the top N hot functions by tottime and ncalls (default N: 25)",
)
args = parser.parse_args(argv)
warnings.filterwarnings("ignore")
@@ -127,6 +170,10 @@ def main(argv: list[str] | None = None) -> int:
print(f"numba: {'ACTIVE - JIT speedups apply' if numba else 'ABSENT - @njit falls back to plain Python'}")
print(f"rows: {args.rows} repeats: {args.repeats} indicators: {len(names)}\n")
if args.profile is not None:
profile(names, args.rows, args.profile)
return 0
results = bench(names, args.rows, args.repeats)
ok = [r for r in results if r["status"] == "ok"]
ok.sort(key=lambda r: r["native_ms"], reverse=True)