# 示例代码
## 交易策略示例
```python
import techkit as tk
import numpy as np
def simple_strategy(prices: np.ndarray) -> list:
"""Simple RSI + MA crossover strategy."""
sma_fast = tk.SMA(10).calculate(prices)
sma_slow = tk.SMA(30).calculate(prices)
rsi = tk.RSI(14).calculate(prices)
signals = []
for i in range(1, len(prices)):
if np.isnan(sma_fast[i]) or np.isnan(sma_slow[i]) or np.isnan(rsi[i]):
signals.append(0)
continue
# Buy: Fast MA crosses above Slow MA, RSI < 70
if sma_fast[i] > sma_slow[i] and sma_fast[i-1] <= sma_slow[i-1] and rsi[i] < 70:
signals.append(1)
# Sell: Fast MA crosses below Slow MA, RSI > 30
elif sma_fast[i] < sma_slow[i] and sma_fast[i-1] >= sma_slow[i-1] and rsi[i] > 30:
signals.append(-1)
else:
signals.append(0)
return signals
```
## 实时流式处理
```python
import techkit as tk
import asyncio
class IndicatorManager:
def __init__(self):
self.sma = tk.SMA(20)
self.rsi = tk.RSI(14)
self.macd = tk.MACD(12, 26, 9)
self.bbands = tk.BBANDS(20, 2.0)
def on_price(self, price: float) -> dict:
return {
'sma': self.sma.update(price),
'rsi': self.rsi.update(price),
'macd': self.macd.update(price),
'bbands': self.bbands.update(price),
}
async def stream_processor():
manager = IndicatorManager()
async for price in price_stream():
indicators = manager.on_price(price)
if indicators['rsi'].valid:
rsi_val = indicators['rsi'].value
if rsi_val > 70:
print(f"⚠️ Overbought: RSI = {rsi_val:.1f}")
elif rsi_val < 30:
print(f"⚠️ Oversold: RSI = {rsi_val:.1f}")
```
## Pandas 集成
```python
import techkit as tk
import pandas as pd
import numpy as np
def add_indicators(df: pd.DataFrame) -> pd.DataFrame:
"""Add technical indicators to DataFrame."""
close = df['close'].values
df['sma_20'] = tk.SMA(20).calculate(close)
df['sma_50'] = tk.SMA(50).calculate(close)
df['rsi_14'] = tk.RSI(14).calculate(close)
macd, signal, hist = tk.MACD(12, 26, 9).calculate(close)
df['macd'] = macd
df['macd_signal'] = signal
df['macd_hist'] = hist
upper, middle, lower = tk.BBANDS(20, 2.0).calculate(close)
df['bb_upper'] = upper
df['bb_middle'] = middle
df['bb_lower'] = lower
# OHLCV indicators
df['atr_14'] = tk.ATR(14).calculate_ohlcv(
df['open'].values, df['high'].values,
df['low'].values, df['close'].values,
df['volume'].values
)
return df
```
## 浏览器示例
```html
TechKit Browser Demo
```