5.2 KiB
5.2 KiB
In [ ]:
import numpy as np
from ferro_ta import BBANDS, EMA, MACD, RSI, SMA
# Synthetic OHLCV data
np.random.seed(42)
n = 200
close = np.cumprod(1 + np.random.randn(n) * 0.01) * 100
high = close * (1 + np.abs(np.random.randn(n)) * 0.005)
low = close * (1 - np.abs(np.random.randn(n)) * 0.005)
volume = np.random.randint(1_000, 10_000, n).astype(float)
print(f"Generated {n} bars")In [ ]:
sma_20 = SMA(close, timeperiod=20)
ema_20 = EMA(close, timeperiod=20)
print("SMA(20):", sma_20[-5:])
print("EMA(20):", ema_20[-5:])In [ ]:
rsi = RSI(close, timeperiod=14)
print("RSI(14):", rsi[-5:])
print(f"RSI range: [{np.nanmin(rsi):.2f}, {np.nanmax(rsi):.2f}]")In [ ]:
macd_line, signal, hist = MACD(close, fastperiod=12, slowperiod=26, signalperiod=9)
print("MACD line: ", macd_line[-5:])
print("Signal: ", signal[-5:])
print("Histogram: ", hist[-5:])In [ ]:
upper, middle, lower = BBANDS(close, timeperiod=20, nbdevup=2.0, nbdevdn=2.0)
print("Upper band: ", upper[-5:])
print("Middle band:", middle[-5:])
print("Lower band: ", lower[-5:])In [ ]:
from ferro_ta.batch import batch_rsi, batch_sma
# Simulate 5 symbols
data = np.random.default_rng(0).random((200, 5)) * 100 + 50
sma_result = batch_sma(data, timeperiod=20)
rsi_result = batch_rsi(data, timeperiod=14)
print("Batch SMA shape:", sma_result.shape) # (200, 5)
print("Batch RSI shape:", rsi_result.shape) # (200, 5)In [ ]:
from ferro_ta.pipeline import Pipeline
pipe = (
Pipeline()
.add("sma_20", SMA, timeperiod=20)
.add("ema_20", EMA, timeperiod=20)
.add("rsi_14", RSI, timeperiod=14)
.add(
"bb",
BBANDS,
output_keys=["bb_upper", "bb_mid", "bb_lower"],
timeperiod=20,
nbdevup=2.0,
nbdevdn=2.0,
)
)
results = pipe.run(close)
print("Pipeline outputs:", list(results.keys()))
print("SMA last 3:", results["sma_20"][-3:])In [ ]:
try:
import pandas as pd
s = pd.Series(close, name="close")
sma_pd = SMA(s, timeperiod=20)
print("Result type:", type(sma_pd)) # pandas.Series
print("Index preserved:", list(sma_pd.index[:3]))
except ImportError:
print("pandas not installed")In [ ]:
from ferro_ta import FerroTAValueError
from ferro_ta.exceptions import check_timeperiod
try:
check_timeperiod(0)
except FerroTAValueError as e:
print("Caught:", e)