85 lines
2.8 KiB
Python
85 lines
2.8 KiB
Python
"""
|
||
负向指标 (5维) — calculate_negative_features(df)
|
||
注意: #53 trend_consistency_20d 与 #39 同名不同义，更新字典时会覆盖 #39
|
||
"""
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
|
||
def calculate_negative_features(ctx) -> pd.DataFrame:
|
||
"""计算负向指标 (5维)"""
|
||
kline = ctx.kline.copy()
|
||
if kline.empty:
|
||
return pd.DataFrame()
|
||
|
||
kline = kline.sort_values(['stock_code', 'trade_date'])
|
||
candidates = ctx.candidates
|
||
|
||
features = {}
|
||
grouped = kline.groupby('stock_code')
|
||
|
||
for code, group in grouped:
|
||
if code not in candidates:
|
||
continue
|
||
g = group.tail(120)
|
||
if len(g) < 20:
|
||
continue
|
||
|
||
closes = g['close'].values
|
||
opens = g['open'].values
|
||
highs = g['high'].values
|
||
lows = g['low'].values
|
||
volumes = g['volume'].values
|
||
|
||
feat = {}
|
||
|
||
# 53. trend_consistency_20d (负向版本): |mean(return>0) - 0.5| × 100
|
||
# 衡量偏离均衡的程度，越接近50%越弱
|
||
rets_20 = np.diff(closes[-21:]) / np.where(closes[-21:-1] > 0, closes[-21:-1], 1)
|
||
feat['trend_consistency_20d'] = abs(np.mean(rets_20 > 0) - 0.5) * 100
|
||
|
||
# 54. max_consecutive_direction_20d: 最大连续同向天数
|
||
rets_sign = np.sign(np.diff(closes[-21:]))
|
||
max_consec = 0
|
||
curr_consec = 0
|
||
curr_sign = 0
|
||
for s in rets_sign:
|
||
if s != 0 and s == curr_sign:
|
||
curr_consec += 1
|
||
elif s != 0:
|
||
curr_sign = s
|
||
curr_consec = 1
|
||
else:
|
||
curr_consec = 0
|
||
max_consec = max(max_consec, curr_consec)
|
||
feat['max_consecutive_direction_20d'] = max_consec
|
||
|
||
# 55. gap_risk_20d: mean(|open_t - close_{t-1}|/close_{t-1} > 0.02) × 100
|
||
gap_count = 0
|
||
n = 0
|
||
for i in range(max(0, len(g) - 20), len(g)):
|
||
if i > 0 and closes[i - 1] > 0:
|
||
gap_pct = abs(opens[i] - closes[i - 1]) / closes[i - 1]
|
||
if gap_pct > 0.02:
|
||
gap_count += 1
|
||
n += 1
|
||
feat['gap_risk_20d'] = gap_count / n * 100 if n > 0 else 0
|
||
|
||
# 56. liquidity_drying_up_20d: min(vol_20d)/mean(vol_60d)
|
||
vol_20 = volumes[-20:]
|
||
vol_60 = volumes[-60:] if len(volumes) >= 60 else volumes
|
||
feat['liquidity_drying_up_20d'] = (
|
||
np.min(vol_20) / np.mean(vol_60) if np.mean(vol_60) > 0 else 1
|
||
)
|
||
|
||
# 57. price_stagnation_20d: (max(high_20d)-min(low_20d))/close × 100
|
||
h20 = np.max(highs[-20:])
|
||
l20 = np.min(lows[-20:])
|
||
feat['price_stagnation_20d'] = (
|
||
(h20 - l20) / closes[-1] * 100 if closes[-1] > 0 else 0
|
||
)
|
||
|
||
features[code] = feat
|
||
|
||
return pd.DataFrame.from_dict(features, orient='index')
|