Pratik Bhadane
285 lines
8.9 KiB
Python
285 lines
8.9 KiB
Python
"""
|
||
ferro_ta.tools — Stable Tool Wrappers for Agent / LLM Integration
|
||
=================================================================
|
||
|
||
Provides stable, well-documented functions that are easy to wrap as
|
||
LangChain/LlamaIndex/OpenAI Function tools or to call from automated agents.
|
||
|
||
All functions have clear signatures, descriptive docstrings, and return
|
||
JSON-serializable types so that agent frameworks can inspect and call them
|
||
without special handling.
|
||
|
||
See ``docs/agentic.md`` for the full agentic workflow guide, LangChain
|
||
integration examples, and scheduling instructions.
|
||
|
||
Quick start
|
||
-----------
|
||
>>> import numpy as np
|
||
>>> from ferro_ta.tools import compute_indicator, run_backtest, list_indicators
|
||
>>>
|
||
>>> close = np.cumprod(1 + np.random.default_rng(0).normal(0, 0.01, 100)) * 100
|
||
>>>
|
||
>>> # Compute a single indicator by name
|
||
>>> result = compute_indicator("SMA", close, timeperiod=14)
|
||
>>>
|
||
>>> # Run a backtest
|
||
>>> summary = run_backtest("rsi_30_70", close)
|
||
>>> print(summary["final_equity"])
|
||
|
||
API
|
||
---
|
||
compute_indicator(name, *args, **kwargs) → array or dict
|
||
Compute a built-in or registered indicator by name.
|
||
|
||
run_backtest(strategy, close, **kwargs) → dict
|
||
Run a backtest and return a summary dict.
|
||
|
||
list_indicators() → list[str]
|
||
list all registered indicator names.
|
||
|
||
describe_indicator(name) → str
|
||
Return the docstring of a registered indicator (or a summary).
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any, Union
|
||
|
||
import numpy as np
|
||
from numpy.typing import ArrayLike, NDArray
|
||
|
||
__all__ = [
|
||
"compute_indicator",
|
||
"run_backtest",
|
||
"list_indicators",
|
||
"describe_indicator",
|
||
]
|
||
|
||
|
||
def compute_indicator(
|
||
name: str,
|
||
*args: ArrayLike,
|
||
**kwargs: Any,
|
||
) -> Union[NDArray[np.float64], dict[str, NDArray[np.float64]]]:
|
||
"""Compute a named indicator and return the result.
|
||
|
||
Delegates to the ferro_ta registry so that both built-in and custom
|
||
indicators can be called by name.
|
||
|
||
Parameters
|
||
----------
|
||
name : str
|
||
Indicator name (e.g. ``"SMA"``, ``"RSI"``, ``"BBANDS"``).
|
||
Case-sensitive; use :func:`list_indicators` to see all names.
|
||
*args : array-like
|
||
Positional data arrays forwarded to the indicator (e.g. close, high).
|
||
**kwargs
|
||
Parameter keyword arguments forwarded to the indicator
|
||
(e.g. ``timeperiod=14``).
|
||
|
||
Returns
|
||
-------
|
||
ndarray or dict of str → ndarray
|
||
For single-output indicators, returns a 1-D ``numpy.ndarray``.
|
||
For multi-output indicators (e.g. BBANDS, MACD), returns a dict
|
||
mapping output names to arrays. The dict keys follow TA-Lib
|
||
conventions where known (``"upper"``/``"middle"``/``"lower"`` for
|
||
BBANDS; ``"macd"``/``"signal"``/``"hist"`` for MACD; etc.).
|
||
|
||
Raises
|
||
------
|
||
ferro_ta.registry.FerroTARegistryError
|
||
If *name* is not a known indicator.
|
||
|
||
Examples
|
||
--------
|
||
>>> import numpy as np
|
||
>>> from ferro_ta.tools import compute_indicator
|
||
>>> close = np.linspace(100, 110, 20)
|
||
>>> result = compute_indicator("SMA", close, timeperiod=5)
|
||
>>> result.shape
|
||
(20,)
|
||
>>> bb = compute_indicator("BBANDS", close, timeperiod=5)
|
||
>>> sorted(bb.keys())
|
||
['lower', 'middle', 'upper']
|
||
"""
|
||
from ferro_ta.core.registry import run as _registry_run
|
||
|
||
raw = _registry_run(name, *args, **kwargs)
|
||
|
||
if isinstance(raw, tuple):
|
||
# Multi-output: try to map to named keys for well-known indicators
|
||
_multi_keys: dict[str, list[str]] = {
|
||
"BBANDS": ["upper", "middle", "lower"],
|
||
"MACD": ["macd", "signal", "hist"],
|
||
"MACDEXT": ["macd", "signal", "hist"],
|
||
"MACDFIX": ["macd", "signal", "hist"],
|
||
"STOCH": ["slowk", "slowd"],
|
||
"STOCHF": ["fastk", "fastd"],
|
||
"STOCHRSI": ["fastk", "fastd"],
|
||
"AROON": ["aroondown", "aroonup"],
|
||
"HT_PHASOR": ["inphase", "quadrature"],
|
||
"HT_SINE": ["sine", "leadsine"],
|
||
"MAMA": ["mama", "fama"],
|
||
}
|
||
keys = _multi_keys.get(name.upper())
|
||
if keys and len(keys) == len(raw):
|
||
return {k: np.asarray(v, dtype=np.float64) for k, v in zip(keys, raw)}
|
||
# Fallback: use integer keys
|
||
return {str(i): np.asarray(v, dtype=np.float64) for i, v in enumerate(raw)}
|
||
|
||
return np.asarray(raw, dtype=np.float64)
|
||
|
||
|
||
def run_backtest(
|
||
strategy: str,
|
||
close: ArrayLike,
|
||
commission_per_trade: float = 0.0,
|
||
slippage_bps: float = 0.0,
|
||
**strategy_kwargs: Any,
|
||
) -> dict[str, Any]:
|
||
"""Run a named backtest strategy and return a summary dictionary.
|
||
|
||
This is a convenience wrapper around :func:`ferro_ta.backtest.backtest`
|
||
that returns a JSON-serializable summary dict rather than a
|
||
``BacktestResult`` object, making it easy to use from agent tools.
|
||
|
||
Parameters
|
||
----------
|
||
strategy : str
|
||
Name of the built-in strategy: ``"rsi_30_70"``, ``"sma_crossover"``,
|
||
or ``"macd_crossover"``.
|
||
close : array-like
|
||
Close prices (1-D, at least 2 bars).
|
||
commission_per_trade : float
|
||
Fixed commission deducted from equity on each position change.
|
||
slippage_bps : float
|
||
Slippage in basis points applied on position-change bars.
|
||
**strategy_kwargs
|
||
Extra kwargs forwarded to the strategy function
|
||
(e.g. ``timeperiod=14``, ``oversold=25``).
|
||
|
||
Returns
|
||
-------
|
||
dict
|
||
Summary with the following keys:
|
||
|
||
* ``"strategy"`` — the strategy name used.
|
||
* ``"n_bars"`` — number of price bars.
|
||
* ``"n_trades"`` — number of position changes.
|
||
* ``"final_equity"`` — terminal equity value (start = 1.0).
|
||
* ``"max_drawdown"`` — maximum drawdown fraction (0–1, positive value
|
||
represents the magnitude of loss).
|
||
* ``"equity"`` — equity curve as a Python list of floats.
|
||
* ``"signals"`` — signal array as a Python list.
|
||
|
||
Examples
|
||
--------
|
||
>>> import numpy as np
|
||
>>> from ferro_ta.tools import run_backtest
|
||
>>> close = np.cumprod(1 + np.random.default_rng(0).normal(0, 0.01, 100)) * 100
|
||
>>> summary = run_backtest("rsi_30_70", close)
|
||
>>> isinstance(summary["final_equity"], float)
|
||
True
|
||
"""
|
||
from ferro_ta.analysis.backtest import backtest as _backtest
|
||
|
||
result = _backtest(
|
||
close,
|
||
strategy=strategy,
|
||
commission_per_trade=commission_per_trade,
|
||
slippage_bps=slippage_bps,
|
||
**strategy_kwargs,
|
||
)
|
||
|
||
equity = np.asarray(result.equity, dtype=np.float64)
|
||
# Compute max drawdown
|
||
running_max = np.maximum.accumulate(equity)
|
||
drawdowns = (running_max - equity) / np.where(running_max > 0, running_max, 1.0)
|
||
max_dd = float(np.nanmax(drawdowns)) if len(drawdowns) > 0 else 0.0
|
||
|
||
return {
|
||
"strategy": strategy,
|
||
"n_bars": len(result.signals),
|
||
"n_trades": result.n_trades,
|
||
"final_equity": result.final_equity,
|
||
"max_drawdown": max_dd,
|
||
"equity": equity.tolist(),
|
||
"signals": np.asarray(result.signals, dtype=np.float64).tolist(),
|
||
}
|
||
|
||
|
||
def list_indicators() -> list[str]:
|
||
"""Return a sorted list of all registered indicator names.
|
||
|
||
Includes both built-in ferro_ta indicators and any custom indicators
|
||
registered via :func:`ferro_ta.registry.register`.
|
||
|
||
Returns
|
||
-------
|
||
list of str
|
||
Sorted list of indicator names (e.g. ``["AD", "ADOSC", "ADX", …]``).
|
||
|
||
Examples
|
||
--------
|
||
>>> from ferro_ta.tools import list_indicators
|
||
>>> names = list_indicators()
|
||
>>> "SMA" in names
|
||
True
|
||
>>> "RSI" in names
|
||
True
|
||
"""
|
||
from ferro_ta.core.registry import list_indicators as _list
|
||
|
||
return _list()
|
||
|
||
|
||
def describe_indicator(name: str) -> str:
|
||
"""Return a human-readable description of a registered indicator.
|
||
|
||
Looks up the indicator's docstring and returns the first paragraph (up to
|
||
the first blank line) so it can be used in agent prompts or tool
|
||
descriptions.
|
||
|
||
Parameters
|
||
----------
|
||
name : str
|
||
Indicator name (case-sensitive). Use :func:`list_indicators` to get
|
||
valid names.
|
||
|
||
Returns
|
||
-------
|
||
str
|
||
The first paragraph of the indicator's docstring, or a fallback
|
||
message if no docstring is available.
|
||
|
||
Raises
|
||
------
|
||
ferro_ta.registry.FerroTARegistryError
|
||
If *name* is not a known indicator.
|
||
|
||
Examples
|
||
--------
|
||
>>> from ferro_ta.tools import describe_indicator
|
||
>>> desc = describe_indicator("SMA")
|
||
>>> isinstance(desc, str) and len(desc) > 0
|
||
True
|
||
"""
|
||
from ferro_ta.core.registry import get as _get
|
||
|
||
func = _get(name)
|
||
doc = getattr(func, "__doc__", None) or ""
|
||
if not doc.strip():
|
||
return f"{name}: no description available."
|
||
|
||
# Return only the first paragraph (before the first blank line)
|
||
lines = doc.strip().splitlines()
|
||
para: list[str] = []
|
||
for line in lines:
|
||
stripped = line.strip()
|
||
if stripped == "" and para:
|
||
break
|
||
para.append(stripped)
|
||
|
||
return " ".join(para).strip() or f"{name}: no description available."
|