技能创建器
skill-creator
创建、改进和评估自定义技能,把专业流程转化为可重复使用的 AI 能力。
technical-momentum-indicators【数据与研究】"Implements rsi, macd, stochastic oscillators and momentum analysis for risk management and algorithmic trading execution."
独立 GitHub 快照 · 自动分类 · 以原作者文档为准
name: technical-momentum-indicators compatibility: opencode completeness: 95 content-types:
Role: Measure the speed and strength of price movement for timing entries and exits
Philosophy: Momentum leads price; divergences and overbought/oversold conditions signal potential reversals
Before completing your task, verify:
Relative paths in this skill (e.g., scripts/, reference/) are relative to this base directory.
import numpy as np
import pandas as pd
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
from scipy import stats
@dataclass
class MomentumSignal:
"""Momentum oscillator signal."""
oscillator: str
value: float
signal_type: str # 'overbought', 'oversold', 'cross', 'divergence'
strength: float # 0-1
timeframe: str
class MomentumAnalyzer:
"""Analyzes momentum using multiple oscillators."""
def __init__(self):
self.overbought = 0.7
self.oversold = 0.3
def calculate_rsi(
self, prices: np.ndarray, period: int = 14
) -> Tuple[np.ndarray, np.ndarray]:
"""Calculate RSI and overbought/oversold zones."""
if len(prices) < period + 1:
return np.array([50] * len(prices)), np.array([0.7] * len(prices))
deltas = np.diff(prices)
gains = np.where(deltas > 0, deltas, 0)
losses = np.where(deltas < 0, -deltas, 0)
avg_gain = np.zeros(len(prices))
avg_loss = np.zeros(len(prices))
avg_gain[period] = np.mean(gains[:period])
avg_loss[period] = np.mean(losses[:period])
for i in range(period + 1, len(prices)):
avg_gain[i] = (avg_gain[i-1] * (period - 1) + gains[i-1]) / period
avg_loss[i] = (avg_loss[i-1] * (period - 1) + losses[i-1]) / period
rs = avg_gain / (avg_loss + 1e-8)
rsi = 100 - (100 / (1 + rs))
# Normalize to 0-1 scale
normalized_rsi = rsi / 100
return normalized_rsi, avg_gain, avg_loss
def calculate_macd(
self, prices: np.ndarray, fast: int = 12, slow: int = 26, signal: int = 9
) -> Dict[str, np.ndarray]:
"""Calculate MACD, signal line, and histogram."""
if len(prices) < slow + signal:
return {'macd': np.array([0]), 'signal': np.array([0]), 'histogram': np.array([0])}
ema_fast = self._calculate_ema(prices, fast)
ema_slow = self._calculate_ema(prices, slow)
macd_line = ema_fast - ema_slow
# Signal line is EMA of MACD
signal_line = self._calculate_ema(macd_line, signal)
histogram = macd_line - signal_line
return {
'macd': macd_line,
'signal': signal_line,
'histogram': histogram
}
def calculate_stochastic(
self, high: np.ndarray, low: np.ndarray, close: np.ndarray,
period: int = 14, k_smooth: int = 3
) -> Dict[str, np.ndarray]:
"""Calculate Stochastic Oscillator."""
if len(high) < period:
return {'k': np.array([50]), 'd': np.array([50])}
# Calculate %K
highest_high = np.maximum.rolling(high, period)
lowest_low = np.minimum.rolling(low, period)
k = 100 * (close - lowest_low) / (highest_high - lowest_low + 1e-8)
k = np.nan_to_num(k, nan=50)
# Calculate %D (smoothed %K)
d = np.convolve(k, np.ones(k_smooth)/k_smooth, mode='same')
d = np.nan_to_num(d, nan=50)
return {'k': k / 100, 'd': d / 100} # Normalize to 0-1
def calculate_roc(self, prices: np.ndarray, period: int = 12) -> np.ndarray:
"""Calculate Rate of Change."""
if len(prices) < period:
return np.array([0] * len(prices))
roc = (prices - np.roll(prices, period)) / (np.roll(prices, period) + 1e-8) * 100
roc[:period] = 0
# Normalize to 0-1
roc_normalized = 1 / (1 + np.exp(-roc / 50)) # Sigmoid normalization
return roc_normalized
def detect_oscillator_divergence(
self, prices: np.ndarray, oscillator: np.ndarray,
lookback: int = 20
) -> List[Dict]:
"""Detect regular and hidden divergences."""
divergences = []
for i in range(lookback, len(prices)):
# Regular divergence (reversal signal)
if (prices[i] > prices[i-lookback] and
oscillator[i] < oscillator[i-lookback] and
oscillator[i] > 0.7): # Overbought
divergences.append({
'type': 'bearish_regular',
'price_trend': 'up',
'oscillator_trend': 'down',
'strength': abs(prices[i] - prices[i-lookback]) / prices[i-lookback]
})
if (prices[i] < prices[i-lookback] and
oscillator[i] > oscillator[i-lookback] and
oscillator[i] < 0.3): # Oversold
divergences.append({
'type': 'bullish_regular',
'price_trend': 'down',
'oscillator_trend': 'up',
'strength': abs(prices[i] - prices[i-lookback]) / prices[i-lookback]
})
# Hidden divergence (continuation signal)
if (prices[i] < prices[i-lookback] and
oscillator[i] > oscillator[i-lookback] and
oscillator[i] < 0.3):
divergences.append({
'type': 'bullish_hidden',
'price_trend': 'down',
'oscillator_trend': 'up',
'strength': abs(oscillator[i] - oscillator[i-lookback])
})
if (prices[i] > prices[i-lookback] and
oscillator[i] < oscillator[i-lookback] and
oscillator[i] > 0.7):
divergences.append({
'type': 'bearish_hidden',
'price_trend': 'up',
'oscillator_trend': 'down',
'strength': abs(oscillator[i] - oscillator[i-lookback])
})
return divergences
def _calculate_ema(self, data: np.ndarray, period: int) -> np.ndarray:
"""Calculate Exponential Moving Average."""
if len(data) < period:
return data
ema = np.zeros(len(data))
ema[:period] = np.mean(data[:period])
multiplier = 2 / (period + 1)
for i in range(period, len(data)):
ema[i] = (data[i] - ema[i-1]) * multiplier + ema[i-1]
return ema
from __future__ import annotations
import logging
from dataclasses import dataclass
from typing import Optional
logger = logging.getLogger(__name__)
@dataclass(frozen=True)
class TradeSignal:
"""Immutable trade signal with all required validation constraints."""
symbol: str
side: str # "buy" or "sell"
price: float
quantity: float
confidence: float # 0.0 to 1.0
reason: str
def validate(self) -> bool:
"""Validate that the trade signal meets all business constraints."""
if self.quantity <= 0:
raise ValueError(f"Quantity must be positive, got {self.quantity}")
if self.price <= 0:
raise ValueError(f"Price must be positive, got {self.price}")
if not 0.0 <= self.confidence <= 1.0:
raise ValueError(f"Confidence must be between 0 and 1, got {self.confidence}")
return True
def generate_trade_signal(
symbol: str,
side: str,
price: float,
quantity: float,
confidence: float,
reason: str,
) -> TradeSignal:
"""Generate a validated trade signal with guard clause checks."""
if side not in ("buy", "sell"):
raise ValueError(f"Invalid side '{side}', must be 'buy' or 'sell'")
signal = TradeSignal(
symbol=symbol,
side=side,
price=price,
quantity=quantity,
confidence=confidence,
reason=reason,
)
signal.validate()
logger.info("Trade signal generated: %s %s %.4f @ %.2f (confidence=%.2f)",
symbol, side, quantity, price, confidence)
return signal
def execute_with_risk_check(signal: TradeSignal, max_position_pct: float = 0.05) -> dict:
"""Execute a trade signal after applying risk management checks."""
adjusted_quantity = signal.quantity
if signal.side == "buy" and signal.quantity > max_position_pct:
logger.warning("Position %s exceeds max %.1f%% — capping to %.4f",
signal.symbol, max_position_pct * 100, max_position_pct)
adjusted_quantity = max_position_pct
return {
"symbol": signal.symbol,
"side": signal.side,
"price": signal.price,
"quantity": adjusted_quantity,
"capped": adjusted_quantity < signal.quantity,
"confidence": signal.confidence,
"status": "submitted",
}
Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.