dekun
102 lines
3.8 KiB
Python
102 lines
3.8 KiB
Python
from __future__ import annotations
|
||
|
||
import math
|
||
from statistics import mean
|
||
|
||
from .candle_rows import rows_to_ohlcv
|
||
|
||
|
||
def build_daily_programmatic(rows_1d: list[list[str]], est_quote_vol_24h_usdt: float) -> dict:
|
||
"""
|
||
日线程序化特征：上方空间（距阶段高）、成交活跃度、简单阻力代理（现价与区间高之间局部高点数量）。
|
||
"""
|
||
_, high, low, close, vol = rows_to_ohlcv(rows_1d)
|
||
if len(close) < 10:
|
||
return {"error": "insufficient_daily", "have": len(close)}
|
||
|
||
last = close[-1]
|
||
look = min(60, len(close))
|
||
hi = max(high[-look:])
|
||
lo = min(low[-look:])
|
||
mid = (hi + lo) / 2 if hi > lo else last
|
||
range_pct = ((hi - lo) / mid) * 100 if mid > 0 else 0.0
|
||
upside_pct = ((hi - last) / last) * 100 if last > 0 else 0.0
|
||
|
||
# 现价上方到区间高：统计「局部高点」数量作为中间阻力代理（越多越密）
|
||
seg_h = high[-look:]
|
||
seg_l = low[-look:]
|
||
local_peaks = 0
|
||
for i in range(1, len(seg_h) - 1):
|
||
if seg_h[i] >= seg_h[i - 1] and seg_h[i] >= seg_h[i + 1]:
|
||
if seg_h[i] > last * 1.002 and seg_h[i] < hi * 0.998:
|
||
local_peaks += 1
|
||
|
||
vol_tail = vol[-20:] if len(vol) >= 20 else vol
|
||
vol_mean = mean(vol_tail[:-1]) if len(vol_tail) > 1 else (vol_tail[0] if vol_tail else 1.0)
|
||
vol_ratio = (vol_tail[-1] / vol_mean) if vol_mean > 0 else 0.0
|
||
|
||
sma20 = mean(close[-20:]) if len(close) >= 20 else mean(close)
|
||
structure_hint = "price_above_sma20" if last >= sma20 else "price_below_sma20"
|
||
|
||
return {
|
||
"last_close": round(last, 8),
|
||
"range_60d_high": round(hi, 8),
|
||
"range_60d_low": round(lo, 8),
|
||
"range_pct_lookback": round(range_pct, 4),
|
||
"upside_to_range_high_pct": round(max(0.0, upside_pct), 4),
|
||
"mid_resistance_proxy_peaks": local_peaks,
|
||
"volume_last_vs_20d_mean": round(vol_ratio, 4),
|
||
"est_quote_vol_24h_usdt": round(est_quote_vol_24h_usdt, 2),
|
||
"structure_hint": structure_hint,
|
||
"sma20": round(sma20, 8),
|
||
}
|
||
|
||
|
||
def programmatic_scores(prog: dict) -> dict:
|
||
"""归一化子分数 0–100，供合成 composite。"""
|
||
if prog.get("error"):
|
||
return {"vol": 0.0, "upside": 0.0, "liquidity": 0.0, "mid_clear": 0.0}
|
||
|
||
est = float(prog.get("est_quote_vol_24h_usdt") or 0.0)
|
||
# 成交额：10M≈35，100M≈70
|
||
vol_score = min(100.0, max(0.0, math.log10(est / 1e6 + 1) * 32.0))
|
||
|
||
upside = float(prog.get("upside_to_range_high_pct") or 0.0)
|
||
upside_score = min(100.0, upside * 4.0)
|
||
|
||
vr = float(prog.get("volume_last_vs_20d_mean") or 0.0)
|
||
liquidity_score = min(100.0, max(0.0, (vr - 1.0) * 35.0 + 40.0))
|
||
|
||
peaks = int(prog.get("mid_resistance_proxy_peaks") or 0)
|
||
mid_clear_score = max(0.0, 100.0 - peaks * 12.0)
|
||
|
||
return {
|
||
"vol": round(vol_score, 2),
|
||
"upside": round(upside_score, 2),
|
||
"liquidity": round(liquidity_score, 2),
|
||
"mid_clear": round(mid_clear_score, 2),
|
||
}
|
||
|
||
|
||
def composite_score(gemma_priority: float, sub: dict) -> float:
|
||
"""gemma_priority 1–10；与程序化子分合成 0–100。"""
|
||
g = max(1.0, min(10.0, gemma_priority)) * 10.0
|
||
p = 0.35 * g
|
||
p += 0.2 * sub.get("vol", 0.0)
|
||
p += 0.2 * sub.get("upside", 0.0)
|
||
p += 0.15 * sub.get("liquidity", 0.0)
|
||
p += 0.1 * sub.get("mid_clear", 0.0)
|
||
return round(min(100.0, max(0.0, p)), 2)
|
||
|
||
|
||
def daily_ohlc_text_block(rows_1d: list[list[str]], max_lines: int = 24) -> str:
|
||
"""给 LLM 的紧凑 OHLCV 文本（时间正序：旧→新，最后一行为最新）。"""
|
||
rows = rows_1d[-max_lines:] if len(rows_1d) > max_lines else rows_1d
|
||
lines = ["ts,o,h,l,c,vol"]
|
||
for item in rows:
|
||
if len(item) < 6:
|
||
continue
|
||
ts, o, h, l, c, v = item[0], item[1], item[2], item[3], item[4], item[5]
|
||
lines.append(f"{ts},{o},{h},{l},{c},{v}")
|
||
return "\n".join(lines)
|