You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
buffer_platform/app/services/plan_generator.py

725 lines
25 KiB

This file contains ambiguous Unicode characters!

This file contains ambiguous Unicode characters that may be confused with others in your current locale. If your use case is intentional and legitimate, you can safely ignore this warning. Use the Escape button to highlight these characters.

"""
V2 交易计划生成器 - 5维度综合评分 + 多周期共振分析
"""
import json
import logging
from pathlib import Path
from typing import Dict, List, Optional, Tuple
from datetime import datetime
from sqlalchemy.orm import Session
from app.services.cache import get_cached_data
from app.analysis_models import (
ReviewPlanV2, SymbolScoreV2, TradingPlanV2, SectorHeat,
)
logger = logging.getLogger(__name__)
# 品种配置加载
CONFIG_DIR = Path(__file__).resolve().parent.parent.parent / "config"
# 板块分类
SECTOR_MAP = {
"贵金属": ["沪银", "沪金"],
"有色金属": ["沪铜", "沪锌", "沪锡", "沪铅", "沪铝", "铝合金", "沪镍"],
"黑色系": ["焦炭", "焦煤", "螺纹钢", "热卷", "铁矿石"],
"能源": ["原油", "低硫燃油", "燃油"],
"化工": ["PTA", "甲醇", "尿素", "PVC", "乙二醇", "合成橡胶", "烧碱", "纯碱", "玻璃", "橡胶"],
"农产品": ["白糖", "棉花", "棕榈油", "豆粕"],
"新能源": ["氧化铝", "多晶硅", "碳酸锂", "工业硅"],
"股指": ["中证1000"],
"航运": ["集运欧线"],
"橡胶": ["20号胶"],
}
# 权重
WEIGHTS = {
"amplitude": 0.20,
"volume": 0.15,
"change": 0.15,
"trend": 0.35,
"activity": 0.15,
}
def load_symbols_config() -> Dict[str, str]:
"""加载品种配置 {中文名: 合约代码}"""
config_path = CONFIG_DIR / "symbols_config.json"
if not config_path.exists():
logger.warning(f"品种配置文件不存在: {config_path}")
return {}
with open(config_path, "r", encoding="utf-8") as f:
data = json.load(f)
return data.get("futures", {})
def _get_candle_values(candles: List[dict]) -> Tuple[list, list, list, list, list]:
"""从K线列表提取 OHLCV"""
opens, highs, lows, closes, volumes = [], [], [], [], []
for c in candles:
opens.append(float(c.get("open", 0)))
highs.append(float(c.get("high", 0)))
lows.append(float(c.get("low", 0)))
closes.append(float(c.get("close", 0)))
volumes.append(float(c.get("volume", 0)))
return opens, highs, lows, closes, volumes
def _calc_single_period_trend(candles: List[dict]) -> float:
"""
计算单周期趋势分 (-50 ~ +50)
MA排列: 多头+50 / 空头-50
MACD: DIF>DEA +30 / DIF<DEA -30
综合: -50 ~ +50 (归一化后映射)
"""
if len(candles) < 5:
return 0.0
last = candles[-1]
ma10 = last.get("ma10")
ma20 = last.get("ma20")
macd_dif = last.get("macd_dif", 0) or 0
macd_dea = last.get("macd_dea", 0) or 0
score = 0.0
# MA排列
if ma10 and ma20:
if ma10 > ma20:
score += 50
else:
score -= 50
# MACD
if macd_dif > macd_dea:
score += 30
else:
score -= 30
# 归一化到 -50 ~ +50
return max(-50, min(50, score * 50 / 80))
def _calc_trend_score(t60: float, t15: float, t5: float) -> Tuple[float, str, str]:
"""
三周期共振判定
Returns: (趋势总分 0-100, 方向标签, 方向简写)
"""
# 共振判定
if t60 > 20 and t15 > 20 and t5 > 20:
direction = "多头共振"
tag = "强多"
raw = 80 + min(20, (t60 + t15 + t5) / 3 * 20 / 50)
elif t60 < -20 and t15 < -20 and t5 < -20:
direction = "空头共振"
tag = "强空"
raw = 80 + min(20, abs(t60 + t15 + t5) / 3 * 20 / 50)
elif t60 > 0 and t15 > 0:
direction = "偏多震荡"
tag = "偏多"
raw = 50 + min(20, (t60 + t15) / 2 * 20 / 50)
elif t60 < 0 and t15 < 0:
direction = "偏空震荡"
tag = "偏空"
raw = 50 + min(20, abs(t60 + t15) / 2 * 20 / 50)
else:
direction = "多空交织"
tag = "震荡"
raw = max(0, 40 - abs(t60 + t15 + t5) / 3)
return round(raw, 1), direction, tag
def _calc_pivot_points(high: float, low: float, close: float) -> dict:
"""计算枢轴点和支撑/阻力位"""
pivot = (high + low + close) / 3
r1 = 2 * pivot - low
s1 = 2 * pivot - high
r2 = pivot + (high - low)
s2 = pivot - (high - low)
return {"pivot": pivot, "r1": r1, "r2": r2, "s1": s1, "s2": s2}
def _normalize_scores(values: List[float], reverse: bool = False) -> List[float]:
"""将一组值归一化到 0-100"""
if not values:
return []
min_v = min(values)
max_v = max(values)
if max_v == min_v:
return [50.0] * len(values)
result = []
for v in values:
if reverse:
score = (max_v - v) / (max_v - min_v) * 100
else:
score = (v - min_v) / (max_v - min_v) * 100
result.append(round(score, 1))
return result
def _normalize_abs_scores(values: List[float]) -> List[float]:
"""将绝对值归一化到 0-100涨跌幅用绝对值排名"""
abs_values = [abs(v) for v in values]
return _normalize_scores(abs_values)
def generate_plan(db: Session, review_date_str: str, week_day: str) -> dict:
"""
生成V2复盘与交易计划
Returns:
{
"review_date_id": int,
"summary": {...},
"scores": [...],
"plans": [...],
"sectors": [...],
}
"""
symbols_config = load_symbols_config()
if not symbols_config:
raise ValueError("品种配置文件为空")
# 反转映射: 合约代码 -> 中文名
code_to_name = {v: k for k, v in symbols_config.items()}
all_symbols = list(symbols_config.values())
logger.info(f"开始生成交易计划: {review_date_str} ({week_day}), 共 {len(all_symbols)} 个品种")
# 将复盘日期作为数据截止时间(当天 23:59:59
try:
end_time = datetime.strptime(review_date_str, "%Y-%m-%d").replace(hour=23, minute=59, second=59)
except ValueError:
end_time = datetime.now()
# ==================== 第一步: 数据收集 ====================
symbol_data = {}
periods_needed = ["daily", "60min", "15min", "5min"]
for symbol in all_symbols:
try:
data = get_cached_data(
db, symbol, "futures",
periods=periods_needed,
end_time=end_time,
max_candles=100,
)
if data and data.get("timeframes"):
symbol_data[symbol] = data
else:
logger.warning(f"{symbol} 无数据")
except Exception as e:
logger.error(f"获取 {symbol} 数据失败: {e}")
if not symbol_data:
raise ValueError("无法从数据库获取任何品种数据,请先进行数据更新")
logger.info(f"成功获取 {len(symbol_data)}/{len(all_symbols)} 个品种数据")
# ==================== 第二步: 多维度打分 ====================
raw_scores = []
for symbol, data in symbol_data.items():
name = code_to_name.get(symbol, symbol)
tf = data.get("timeframes", {})
daily = tf.get("daily", [])
h1 = tf.get("60min", [])
m15 = tf.get("15min", [])
m5 = tf.get("5min", [])
# 日线数据
if not daily or len(daily) < 2:
continue
_, d_highs, d_lows, d_closes, d_volumes = _get_candle_values(daily)
today_high = d_highs[-1]
today_low = d_lows[-1]
today_close = d_closes[-1]
prev_close = d_closes[-2] if len(d_closes) >= 2 else today_close
today_volume = d_volumes[-1]
# 近5日均量
recent_vols = d_volumes[-6:-1] if len(d_volumes) >= 6 else d_volumes[:-1]
avg_vol_5 = sum(recent_vols) / len(recent_vols) if recent_vols else today_volume
# A. 振幅
amplitude_pct = (today_high - today_low) / today_close * 100 if today_close else 0
# B. 量比
volume_ratio = today_volume / avg_vol_5 if avg_vol_5 > 0 else 1.0
# C. 涨跌幅
change_pct = (today_close - prev_close) / prev_close * 100 if prev_close else 0
# D. 多周期趋势
t60 = _calc_single_period_trend(h1) if h1 else 0
t15 = _calc_single_period_trend(m15) if m15 else 0
t5 = _calc_single_period_trend(m5) if m5 else 0
trend_total, direction, direction_tag = _calc_trend_score(t60, t15, t5)
# E. 活跃度 (5min K线近10根)
if m5 and len(m5) >= 5:
recent_m5 = m5[-10:] if len(m5) >= 10 else m5
_, _, _, m5_closes, m5_vols = _get_candle_values(recent_m5)
# 价格变动率
if m5_closes and m5_closes[0] > 0:
price_change_rate = abs(m5_closes[-1] - m5_closes[0]) / m5_closes[0] * 100
else:
price_change_rate = 0
# 量能活跃度
avg_m5_vol = sum(m5_vols) / len(m5_vols) if m5_vols else 0
vol_activity = min(100, avg_m5_vol / 100) if avg_m5_vol > 0 else 0
price_activity = min(100, price_change_rate * 50)
activity_score = (vol_activity + price_activity) / 2
else:
activity_score = 0
# 关键点位
pivots = _calc_pivot_points(today_high, today_low, today_close)
raw_scores.append({
"symbol": symbol,
"name": name,
"close_price": today_close,
"prev_close": prev_close,
"high_price": today_high,
"low_price": today_low,
"volume": today_volume,
"avg_volume_5": avg_vol_5,
"amplitude_pct": round(amplitude_pct, 2),
"volume_ratio": round(volume_ratio, 2),
"change_pct": round(change_pct, 2),
"amplitude_raw": amplitude_pct,
"volume_raw": volume_ratio,
"change_raw": abs(change_pct),
"trend_score": trend_total,
"trend_60m": round(t60, 1),
"trend_15m": round(t15, 1),
"trend_5m": round(t5, 1),
"direction": direction,
"direction_tag": direction_tag,
"activity_raw": activity_score,
"pivots": pivots,
})
if not raw_scores:
raise ValueError("没有足够的数据进行评分计算")
# ==================== 第三步: 归一化评分 ====================
amp_scores = _normalize_scores([s["amplitude_raw"] for s in raw_scores])
vol_scores = _normalize_scores([s["volume_raw"] for s in raw_scores])
chg_scores = _normalize_abs_scores([s["change_raw"] for s in raw_scores])
act_scores = _normalize_scores([s["activity_raw"] for s in raw_scores])
for i, s in enumerate(raw_scores):
s["amplitude_score"] = amp_scores[i]
s["volume_score"] = vol_scores[i]
s["change_score"] = chg_scores[i]
s["activity_score"] = act_scores[i]
# 综合评分 = 振幅×0.20 + 量能×0.15 + 涨跌幅×0.15 + 趋势×0.35 + 活跃度×0.15
# 趋势分映射: trend_score 0-100 直接使用
s["composite_score"] = round(
amp_scores[i] * WEIGHTS["amplitude"]
+ vol_scores[i] * WEIGHTS["volume"]
+ chg_scores[i] * WEIGHTS["change"]
+ s["trend_score"] * WEIGHTS["trend"]
+ act_scores[i] * WEIGHTS["activity"],
1
)
# 排序
raw_scores.sort(key=lambda x: x["composite_score"], reverse=True)
# ==================== 第四步: 分类 ====================
for rank, s in enumerate(raw_scores, 1):
s["rank"] = rank
score = s["composite_score"]
trend = s["trend_score"]
if score >= 55 and trend >= 50:
s["category"] = "green"
elif score >= 45 or (score >= 40 and 30 <= trend < 50):
s["category"] = "yellow"
else:
s["category"] = "red"
# ==================== 第五步: 生成交易计划 ====================
plans = []
green_items = [s for s in raw_scores if s["category"] == "green"]
for s in green_items:
direction = "long" if s["trend_score"] >= 50 else "short"
pivots = s["pivots"]
if direction == "long":
entry_low = round(pivots["s1"], 2)
entry_high = round(pivots["pivot"], 2)
stop_loss = round(pivots["s2"], 2)
target1 = round(pivots["r1"], 2)
target2 = round(pivots["r2"], 2)
else:
entry_low = round(pivots["pivot"], 2)
entry_high = round(pivots["r1"], 2)
stop_loss = round(pivots["r2"], 2)
target1 = round(pivots["s1"], 2)
target2 = round(pivots["s2"], 2)
plans.append({
"symbol": s["symbol"],
"name": s["name"],
"direction": direction,
"composite_score": s["composite_score"],
"entry_low": entry_low,
"entry_high": entry_high,
"stop_loss": stop_loss,
"target1": target1,
"target2": target2,
"trigger": "回踩支撑企稳 + 5m放量突破",
"amplitude_score": s["amplitude_score"],
"volume_score": s["volume_score"],
"trend_score": s["trend_score"],
"activity_score": s["activity_score"],
"category": s["category"],
})
# ==================== 第六步: 板块热度 ====================
sectors = _calc_sector_heat(raw_scores, code_to_name)
# ==================== 第七步: 汇总 ====================
bull_count = sum(1 for s in raw_scores if s["trend_score"] >= 50)
bear_count = sum(1 for s in raw_scores if s["trend_score"] < 30)
neutral_count = len(raw_scores) - bull_count - bear_count
# 核心结论
top3 = raw_scores[:3]
top_names = [f"{s['symbol']} {s['name']}" for s in top3]
main_direction = "多头" if bull_count > bear_count else "空头" if bear_count > bull_count else "震荡"
core_conclusion = f"{main_direction}格局{'延续' if bull_count > 15 else '分化'}{'贵金属+黑色系共振做多' if bull_count > 20 else '板块轮动明显'}"
# 风险提示
risk_warnings = []
if top3:
top_sym = top3[0]
risk_warnings.append(
f"极值品种严禁追{'' if top_sym['trend_score'] >= 50 else ''}"
f"{top_sym['symbol']}({top_sym['composite_score']}分) "
f"{'周一大概率高开' if top_sym['change_pct'] > 2 else '注意风险控制'}"
)
if bear_count > 0:
bear_syms = [s for s in raw_scores if s["category"] == "red" and s["trend_score"] < 30][:2]
if bear_syms:
names = "/".join(s["symbol"] for s in bear_syms)
risk_warnings.append(f"空头品种谨慎:{names} 趋势偏空,需严格止损")
risk_warnings.append("关注宏观数据和消息面变化,可能影响开盘方向")
if any(s["volume_ratio"] > 2.5 for s in raw_scores):
high_vol = [s for s in raw_scores if s["volume_ratio"] > 2.5]
risk_warnings.append(
f"量能异常品种:{', '.join(s['symbol'] for s in high_vol[:3])} "
f"量比超2.5倍,需确认量能可持续性"
)
return {
"review_date": review_date_str,
"week_day": week_day,
"data_basis": f"{review_date_str} 收盘 ({len(symbol_data)} 品种) | 多周期共振分析",
"core_conclusion": core_conclusion,
"bull_count": bull_count,
"bear_count": bear_count,
"neutral_count": neutral_count,
"opportunity_count": len(green_items),
"risk_warnings": risk_warnings,
"scores": raw_scores,
"plans": plans,
"sectors": sectors,
"symbol_count": len(symbol_data),
}
def _calc_sector_heat(scores: List[dict], code_to_name: dict) -> List[dict]:
"""计算板块热度"""
# 构建 symbol -> score 映射
name_to_score = {}
for s in scores:
name_to_score[s["name"]] = s
sectors = []
for sector_name, members_names in SECTOR_MAP.items():
members = []
for mn in members_names:
if mn in name_to_score:
sc = name_to_score[mn]
members.append({
"symbol": sc["symbol"],
"name": sc["name"],
"score": sc["composite_score"],
"trend": sc["trend_score"],
"change_pct": sc["change_pct"],
})
if not members:
continue
avg_score = round(sum(m["score"] for m in members) / len(members), 1)
avg_trend = round(sum(m["trend"] for m in members) / len(members), 0)
if avg_trend > 20:
direction = "多头"
elif avg_trend < -20:
direction = "空头"
else:
direction = "震荡"
# 热度等级
if avg_score >= 75:
heat = 3
elif avg_score >= 60:
heat = 2
elif avg_score >= 45:
heat = 1
else:
heat = 0
# 龙头
leader = max(members, key=lambda m: m["score"])
sectors.append({
"sector_name": sector_name,
"avg_score": avg_score,
"avg_trend": int(avg_trend),
"direction": direction,
"heat_level": heat,
"leader_symbol": leader["symbol"],
"leader_score": leader["score"],
"members": sorted(members, key=lambda m: m["score"], reverse=True),
})
sectors.sort(key=lambda s: s["avg_score"], reverse=True)
return sectors
def save_plan_to_db(db: Session, plan_data: dict) -> int:
"""将计划数据保存到数据库"""
review_date_str = plan_data["review_date"]
week_day = plan_data["week_day"]
# 清理旧数据
existing = db.query(ReviewPlanV2).filter_by(review_date=review_date_str).first()
if existing:
review_plan_id = existing.id
# 清理关联数据
db.query(SymbolScoreV2).filter_by(review_date_id=review_plan_id).delete()
db.query(TradingPlanV2).filter_by(review_date_id=review_plan_id).delete()
db.query(SectorHeat).filter_by(review_date_id=review_plan_id).delete()
# 更新元数据
existing.week_day = week_day
existing.data_basis = plan_data.get("data_basis")
existing.core_conclusion = plan_data.get("core_conclusion")
existing.bull_count = plan_data.get("bull_count")
existing.bear_count = plan_data.get("bear_count")
existing.neutral_count = plan_data.get("neutral_count")
existing.opportunity_count = plan_data.get("opportunity_count")
existing.risk_warnings = plan_data.get("risk_warnings")
else:
review_plan = ReviewPlanV2(
review_date=review_date_str,
week_day=week_day,
data_basis=plan_data.get("data_basis"),
core_conclusion=plan_data.get("core_conclusion"),
bull_count=plan_data.get("bull_count"),
bear_count=plan_data.get("bear_count"),
neutral_count=plan_data.get("neutral_count"),
opportunity_count=plan_data.get("opportunity_count"),
risk_warnings=plan_data.get("risk_warnings"),
)
db.add(review_plan)
db.flush()
review_plan_id = review_plan.id
# 保存评分数据
for s in plan_data.get("scores", []):
pivots = s.get("pivots", {})
score_record = SymbolScoreV2(
review_date_id=review_plan_id,
symbol=s["symbol"],
name=s["name"],
close_price=s.get("close_price"),
prev_close=s.get("prev_close"),
high_price=s.get("high_price"),
low_price=s.get("low_price"),
volume=s.get("volume"),
avg_volume_5=s.get("avg_volume_5"),
amplitude_score=s.get("amplitude_score"),
volume_score=s.get("volume_score"),
change_score=s.get("change_score"),
trend_score=s.get("trend_score"),
activity_score=s.get("activity_score"),
composite_score=s.get("composite_score"),
amplitude_pct=s.get("amplitude_pct"),
change_pct=s.get("change_pct"),
volume_ratio=s.get("volume_ratio"),
trend_60m=s.get("trend_60m"),
trend_15m=s.get("trend_15m"),
trend_5m=s.get("trend_5m"),
direction=s.get("direction"),
direction_tag=s.get("direction_tag"),
category=s.get("category"),
pivot=pivots.get("pivot"),
r1=pivots.get("r1"),
r2=pivots.get("r2"),
s1=pivots.get("s1"),
s2=pivots.get("s2"),
rank=s.get("rank"),
)
db.add(score_record)
# 保存交易计划
for p in plan_data.get("plans", []):
plan_record = TradingPlanV2(
review_date_id=review_plan_id,
symbol=p["symbol"],
name=p["name"],
direction=p["direction"],
composite_score=p.get("composite_score"),
entry_low=p.get("entry_low"),
entry_high=p.get("entry_high"),
stop_loss=p.get("stop_loss"),
target1=p.get("target1"),
target2=p.get("target2"),
trigger=p.get("trigger"),
amplitude_score=p.get("amplitude_score"),
volume_score=p.get("volume_score"),
trend_score=p.get("trend_score"),
activity_score=p.get("activity_score"),
category=p.get("category"),
)
db.add(plan_record)
# 保存板块热度
for sec in plan_data.get("sectors", []):
sector_record = SectorHeat(
review_date_id=review_plan_id,
sector_name=sec["sector_name"],
avg_score=sec.get("avg_score"),
avg_trend=sec.get("avg_trend"),
direction=sec.get("direction"),
heat_level=sec.get("heat_level"),
leader_symbol=sec.get("leader_symbol"),
leader_score=sec.get("leader_score"),
members=sec.get("members"),
)
db.add(sector_record)
db.commit()
logger.info(f"交易计划已保存: {review_date_str}, ID={review_plan_id}")
return review_plan_id
def get_plan_data(db: Session, review_plan_id: int) -> Optional[dict]:
"""从数据库读取完整的计划数据"""
plan_meta = db.query(ReviewPlanV2).filter_by(id=review_plan_id).first()
if not plan_meta:
return None
scores = db.query(SymbolScoreV2).filter_by(
review_date_id=review_plan_id
).order_by(SymbolScoreV2.composite_score.desc()).all()
plans = db.query(TradingPlanV2).filter_by(
review_date_id=review_plan_id
).order_by(TradingPlanV2.composite_score.desc()).all()
sectors = db.query(SectorHeat).filter_by(
review_date_id=review_plan_id
).order_by(SectorHeat.avg_score.desc()).all()
return {
"meta": {
"id": plan_meta.id,
"review_date": plan_meta.review_date,
"week_day": plan_meta.week_day,
"data_basis": plan_meta.data_basis,
"core_conclusion": plan_meta.core_conclusion,
"bull_count": plan_meta.bull_count,
"bear_count": plan_meta.bear_count,
"neutral_count": plan_meta.neutral_count,
"opportunity_count": plan_meta.opportunity_count,
"risk_warnings": plan_meta.risk_warnings,
"created_at": plan_meta.created_at.isoformat() if plan_meta.created_at else None,
},
"scores": [_score_to_dict(s) for s in scores],
"plans": [_plan_to_dict(p) for p in plans],
"sectors": [_sector_to_dict(s) for s in sectors],
}
def _score_to_dict(s: SymbolScoreV2) -> dict:
return {
"id": s.id,
"symbol": s.symbol,
"name": s.name,
"close_price": s.close_price,
"prev_close": s.prev_close,
"high_price": s.high_price,
"low_price": s.low_price,
"volume": s.volume,
"avg_volume_5": s.avg_volume_5,
"amplitude_score": s.amplitude_score,
"volume_score": s.volume_score,
"change_score": s.change_score,
"trend_score": s.trend_score,
"activity_score": s.activity_score,
"composite_score": s.composite_score,
"amplitude_pct": s.amplitude_pct,
"change_pct": s.change_pct,
"volume_ratio": s.volume_ratio,
"trend_60m": s.trend_60m,
"trend_15m": s.trend_15m,
"trend_5m": s.trend_5m,
"direction": s.direction,
"direction_tag": s.direction_tag,
"category": s.category,
"pivot": s.pivot,
"r1": s.r1,
"r2": s.r2,
"s1": s.s1,
"s2": s.s2,
"rank": s.rank,
}
def _plan_to_dict(p: TradingPlanV2) -> dict:
return {
"id": p.id,
"symbol": p.symbol,
"name": p.name,
"direction": p.direction,
"composite_score": p.composite_score,
"entry_low": p.entry_low,
"entry_high": p.entry_high,
"stop_loss": p.stop_loss,
"target1": p.target1,
"target2": p.target2,
"trigger": p.trigger,
"amplitude_score": p.amplitude_score,
"volume_score": p.volume_score,
"trend_score": p.trend_score,
"activity_score": p.activity_score,
"category": p.category,
}
def _sector_to_dict(s: SectorHeat) -> dict:
return {
"id": s.id,
"sector_name": s.sector_name,
"avg_score": s.avg_score,
"avg_trend": s.avg_trend,
"direction": s.direction,
"heat_level": s.heat_level,
"leader_symbol": s.leader_symbol,
"leader_score": s.leader_score,
"members": s.members,
}