Files
stock/stock-html/services/stock_algorithms.py
T
freedakgmail cf7341fc63 perf: 6项性能与正确性优化
1. get_stock_name: 延迟批量保存缓存(60秒去抖),避免每次新缓存都写文件
2. get_stock_name: 优先从DB stock_realtime_price查name,未命中再调腾讯API
3. compute_comprehensive_score: 单股模式也使用_fund_flow_cache(30分钟TTL)
4. _get_kline_from_local_db: 移除conn.autocommit=True,避免污染连接池事务模式
5. deep_analyze: K线天数从180改为120,与实时检测统一
6. detect_all_signals: 已有指标列时跳过重复calc_all_indicators
2026-07-22 08:00:27 +08:00

1323 lines
51 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""
统一算法模块 — 全部核心算法的唯一定义处(Single Source of Truth
包含:
1. compute_recommend — 统一推荐逻辑(买入/卖出/加仓/观望等,严格遵循suanfa.md)
2. get_kline_data — 获取K线数据并返回 DataFrame(优先本地DB → 阿里云 → 腾讯 → 麦蕊 → AKShare)
3. fetch_kline_rows — 获取K线数据并返回 tuple 行列表(用于写入DB同步)
4. get_latest_price — 获取股票最新价格
5. code_to_market — 股票代码→市场判断(SH/SZ/BJ)
6. 各 API session 管理
7. compute_bull_stage — 牛股阶段识别(底部→起爆→确立→加速→补涨)
8. find_bull_stocks — 从扫描结果中找出潜在牛股
调用方:
- routes/analysis.py → compute_recommend, get_kline_data
- services/scheduler.py → compute_recommend, get_latest_price
- full_signal_scan.py → get_kline_data, API sessions
- sync_kline.py → fetch_kline_rows, API sessions
- routes/market.py → get_kline_data
"""
import threading
from datetime import datetime, timedelta
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from config import Config
# ═══════════════════════════════════════════════
# 1. 股票代码 → 市场 工具函数
# ═══════════════════════════════════════════════
def code_to_market(code):
"""6位股票代码 → 市场代码(SH/SZ/BJ)"""
if code.startswith(('0', '3')):
return 'SZ'
elif code.startswith(('8', '9')):
return 'BJ'
else:
return 'SH'
def is_bj_stock(code):
"""是否是北交所股票"""
return code.startswith(('8', '9'))
def code_to_ali_symbol(code):
"""6位股票代码 → 阿里云API格式(SH600519 / SZ000001 / BJ920720"""
return f'{code_to_market(code)}{code}'
def code_to_tencent_symbol(code):
"""6位股票代码 → 腾讯API格式(sh600519 / sz000001 / bj920720"""
return f'{code_to_market(code).lower()}{code}'
# ═══════════════════════════════════════════════
# 2. API Session 管理(线程安全,连接池复用)
# ═══════════════════════════════════════════════
_ali_session = None
_ali_lock = threading.Lock()
_tencent_session = None
_tencent_lock = threading.Lock()
def get_ali_session():
"""获取阿里云API专用 Session(线程安全,单例)"""
global _ali_session
if _ali_session is None:
with _ali_lock:
if _ali_session is None:
s = requests.Session()
retry = Retry(total=2, backoff_factor=0.3,
status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry,
pool_connections=20, pool_maxsize=20)
s.mount('https://', adapter)
s.headers.update({
'Authorization': f'APPCODE {Config.ALICLOUD_APPCODE}',
'Content-Type': 'application/x-www-form-urlencoded',
})
_ali_session = s
return _ali_session
def get_tencent_session():
"""获取腾讯K线API专用 Session(线程安全,单例)"""
global _tencent_session
if _tencent_session is None:
with _tencent_lock:
if _tencent_session is None:
s = requests.Session()
retry = Retry(total=2, backoff_factor=0.3,
status_forcelist=[500, 502, 503, 504])
adapter = HTTPAdapter(max_retries=retry,
pool_connections=20, pool_maxsize=20)
s.mount('https://', adapter)
s.headers.update({
'User-Agent': ('Mozilla/5.0 (Windows NT 10.0; Win64; x64) '
'AppleWebKit/537.36'),
})
_tencent_session = s
return _tencent_session
# API URL 常量
ALICLOUD_KLINE_URL = Config.ALICLOUD_KLINE_URL
TENCENT_KLINE_URL = 'https://proxy.finance.qq.com/ifzqgtimg/appstock/app/newfqkline/get'
# ═══════════════════════════════════════════════
# 3. K线数据获取 — DataFrame 格式(供信号检测/分析用)
# ═══════════════════════════════════════════════
def get_kline_data(stock_code, days=120, use_local_db=True):
"""
获取K线数据,返回 pandas DataFrame (columns: date, open, high, low, close, volume)
数据源优先级: 本地DB → 阿里云API → 腾讯API → 麦蕊API → AKShare
参数:
stock_code: 6位股票代码
days: 获取天数
use_local_db: 是否优先使用本地DB(全景扫描时为True, 实时分析时可为False)
返回:
DataFrame 或 None
"""
import pandas as pd
# 1. 优先从本地数据库读取
if use_local_db:
df = _get_kline_from_local_db(stock_code, days)
if df is not None:
return df
# 2. 阿里云K线API(沪深最稳定,北交所可能不支持)
if not is_bj_stock(stock_code):
df = _fetch_ali_kline_df(stock_code, days)
if df is not None and len(df) >= 30:
return df
# 3. 腾讯K线API(全市场,含北交所)
df = _fetch_tencent_kline_df(stock_code, days)
if df is not None and len(df) >= 30:
return df
# 4. 麦蕊API
df = _fetch_mairui_kline_df(stock_code, days)
if df is not None and len(df) >= 30:
return df
# 5. AKShare
df = _fetch_akshare_kline_df(stock_code, days)
if df is not None and len(df) >= 30:
return df
return None
def _get_kline_from_local_db(stock_code, days=120):
"""从本地数据库读取K线(最快,毫秒级)"""
import pandas as pd
try:
from db import get_db, put_db
conn = get_db()
if not conn:
return None
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
try:
with conn.cursor() as cur:
cur.execute("""
SELECT trade_date, open, high, low, close, volume
FROM stock_kline_daily
WHERE code = %s AND trade_date >= %s
ORDER BY trade_date
""", (stock_code, start_date))
rows = cur.fetchall()
finally:
put_db(conn)
if rows and len(rows) >= 30:
df = pd.DataFrame(rows, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
df['date'] = df['date'].astype(str)
for col in ('open', 'high', 'low', 'close', 'volume'):
df[col] = df[col].astype(float)
return df
except Exception:
pass
return None
# ---- 线程本地连接(供多线程扫描时使用,避免频繁建连) ----
_thread_local = threading.local()
def get_kline_from_local_db_threaded(stock_code, days=120):
"""多线程扫描专用:使用线程本地连接从本地DB读取K线"""
import pandas as pd
try:
conn = getattr(_thread_local, 'kline_conn', None)
if conn is None or conn.closed:
import psycopg2
conn = psycopg2.connect(
host=Config.DB_HOST, port=Config.DB_PORT,
dbname=Config.DB_NAME, user=Config.DB_USER, password=Config.DB_PASSWORD,
)
conn.autocommit = True
_thread_local.kline_conn = conn
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y-%m-%d')
with conn.cursor() as cur:
cur.execute("""
SELECT trade_date, open, high, low, close, volume
FROM stock_kline_daily
WHERE code = %s AND trade_date >= %s
ORDER BY trade_date
""", (stock_code, start_date))
rows = cur.fetchall()
if rows and len(rows) >= 30:
df = pd.DataFrame(rows, columns=['date', 'open', 'high', 'low', 'close', 'volume'])
df['date'] = df['date'].astype(str)
for col in ('open', 'high', 'low', 'close', 'volume'):
df[col] = df[col].astype(float)
return df
except Exception:
pass
return None
def _fetch_ali_kline_df(stock_code, days=120):
"""阿里云K线API → DataFrame"""
import pandas as pd
try:
session = get_ali_session()
symbol = code_to_ali_symbol(stock_code)
resp = session.post(ALICLOUD_KLINE_URL, data={
'symbol': symbol,
'type': '240',
'limit': str(min(days, 300)),
'ma': '5',
}, timeout=10)
if resp.status_code == 200:
data = resp.json()
if data.get('success') and data.get('data', {}).get('list'):
records = []
for item in data['data']['list']:
day_str = item.get('day', '')
if not day_str or len(day_str) < 10:
continue
records.append({
'date': day_str[:10],
'open': float(item.get('open', 0)),
'high': float(item.get('high', 0)),
'low': float(item.get('low', 0)),
'close': float(item.get('close', 0)),
'volume': float(item.get('volume', 0)),
})
if records:
return pd.DataFrame(records)
except Exception:
pass
return None
def _fetch_tencent_kline_df(stock_code, days=120):
"""腾讯K线API → DataFrame(全市场含北交所)"""
import pandas as pd
try:
session = get_tencent_session()
symbol = code_to_tencent_symbol(stock_code)
start_date = (datetime.now() - timedelta(days=days + 30)).strftime('%Y-%m-%d')
resp = session.get(TENCENT_KLINE_URL, params={
'param': f'{symbol},day,{start_date},,{min(days, 300)},qfq',
}, timeout=15)
if resp.status_code == 200:
data = resp.json()
stock_data = data.get('data', {}).get(symbol, {})
klines = stock_data.get('qfqday') or stock_data.get('day') or []
if klines:
records = []
for item in klines:
if len(item) < 6:
continue
# 腾讯格式: [date, open, close, high, low, volume, ...]
records.append({
'date': item[0][:10],
'open': float(item[1]),
'high': float(item[3]), # high = position 3
'low': float(item[4]), # low = position 4
'close': float(item[2]), # close = position 2
'volume': float(item[5]),
})
if records:
return pd.DataFrame(records)
except Exception:
pass
return None
def _fetch_mairui_kline_df(stock_code, days=120):
"""麦蕊API → DataFrame"""
import pandas as pd
try:
from services.mairui_api import get_kline
result = get_kline(stock_code, period='d', days=days, adjust='f')
if result['success'] and result['data']:
df = pd.DataFrame(result['data'])
df.rename(columns={
'date': 'date', 'open': 'open', 'high': 'high',
'low': 'low', 'close': 'close', 'volume': 'volume',
}, inplace=True)
if len(df) >= 30:
return df
except Exception:
pass
return None
def _fetch_akshare_kline_df(stock_code, days=120):
"""AKShare → DataFrame (支持自动降级到腾讯数据源)"""
import pandas as pd
try:
from utils.data_fetcher import fetch_stock_hist
end_date = datetime.now().strftime('%Y%m%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
df = fetch_stock_hist(
stock_code=stock_code, period='daily',
start_date=start_date, end_date=end_date, adjust='qfq',
)
if df is not None and not df.empty:
df = df.rename(columns={
'日期': 'date', '开盘': 'open', '最高': 'high',
'最低': 'low', '收盘': 'close', '成交量': 'volume',
})
df = df[['date', 'open', 'high', 'low', 'close', 'volume']]
return df
except Exception:
pass
return None
# ═══════════════════════════════════════════════
# 4. K线数据获取 — tuple行格式(供 sync_kline.py 写入DB用)
# ═══════════════════════════════════════════════
def fetch_kline_rows(code, days):
"""
获取K线数据,返回 list of tuple: (code, date, open, high, low, close, volume, amount)
数据源优先级: 阿里云API → 腾讯API → 麦蕊API → AKShare
北交所(8XX/9XX)直接走腾讯API
供 sync_kline.py 同步到本地数据库使用。
"""
bj = is_bj_stock(code)
# 北交所:直接用腾讯API(阿里云/Mairui不支持BJ
if bj:
rows = _fetch_tencent_kline_rows(code, days)
if rows:
return rows
return None
# 1. 首选:阿里云K线API
rows = _fetch_ali_kline_rows(code, days)
if rows:
return rows
# 2. 回退:腾讯API
rows = _fetch_tencent_kline_rows(code, days)
if rows:
return rows
# 3. 回退:Mairui API
rows = _fetch_mairui_kline_rows(code, days)
if rows:
return rows
# 4. 最后回退:AKShare
rows = _fetch_akshare_kline_rows(code, days)
if rows:
return rows
return None
def _fetch_ali_kline_rows(code, days):
"""阿里云K线API → tuple rows"""
try:
session = get_ali_session()
symbol = code_to_ali_symbol(code)
resp = session.post(ALICLOUD_KLINE_URL, data={
'symbol': symbol,
'type': '240',
'limit': str(min(days, 300)),
'ma': '5',
}, timeout=10)
if resp.status_code == 200:
data = resp.json()
if data.get('success') and data.get('data', {}).get('list'):
rows = []
for item in data['data']['list']:
day_str = item.get('day', '')
if not day_str or len(day_str) < 10:
continue
rows.append((
code,
day_str[:10],
float(item.get('open', 0)),
float(item.get('high', 0)),
float(item.get('low', 0)),
float(item.get('close', 0)),
int(item.get('volume', 0)),
float(item.get('amount', 0)),
))
if rows:
return rows
except Exception:
pass
return None
def _fetch_tencent_kline_rows(code, days):
"""腾讯K线API → tuple rows(全市场含北交所)"""
try:
symbol = code_to_tencent_symbol(code)
session = get_tencent_session()
start_date = (datetime.now() - timedelta(days=days + 30)).strftime('%Y-%m-%d')
resp = session.get(TENCENT_KLINE_URL, params={
'param': f'{symbol},day,{start_date},,{min(days, 300)},qfq',
}, timeout=15)
if resp.status_code == 200:
data = resp.json()
stock_data = data.get('data', {}).get(symbol, {})
klines = stock_data.get('qfqday') or stock_data.get('day') or []
if klines:
rows = []
for item in klines:
if len(item) < 6:
continue
date_str = item[0]
if not date_str or len(date_str) < 10:
continue
rows.append((
code,
date_str[:10],
float(item[1]), # open
float(item[3]), # high (position 3)
float(item[4]), # low (position 4)
float(item[2]), # close (position 2)
int(float(item[5])), # volume
float(item[8]) * 10000 if len(item) > 8 and item[8] else 0,
))
if rows:
return rows
except Exception:
pass
return None
def _fetch_mairui_kline_rows(code, days):
"""麦蕊API → tuple rows"""
try:
from services.mairui_api import get_kline
result = get_kline(code, period='d', days=days, adjust='f')
if result['success'] and result['data']:
rows = []
for item in result['data']:
date_str = item.get('date', '')
if not date_str:
continue
rows.append((
code,
date_str,
item.get('open', 0),
item.get('high', 0),
item.get('low', 0),
item.get('close', 0),
int(item.get('volume', 0)),
item.get('amount', 0),
))
if rows:
return rows
except Exception:
pass
return None
def _fetch_akshare_kline_rows(code, days):
"""AKShare → tuple rows (支持自动降级到腾讯数据源)"""
try:
from utils.data_fetcher import fetch_stock_hist
end_date = datetime.now().strftime('%Y%m%d')
start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
df = fetch_stock_hist(
stock_code=code, period='daily',
start_date=start_date, end_date=end_date, adjust='qfq',
)
if df is not None and not df.empty:
rows = []
for _, r in df.iterrows():
rows.append((
code,
str(r['日期']),
float(r['开盘']),
float(r['最高']),
float(r['最低']),
float(r['收盘']),
int(r['成交量']),
float(r.get('成交额', 0)),
))
if rows:
return rows
except Exception:
pass
return None
# ═══════════════════════════════════════════════
# 5. 统一推荐算法
# ═══════════════════════════════════════════════
def compute_recommend(signal_status, indicators, triggered_count, is_holding):
"""
统一推荐逻辑 — 全局唯一定义(严格遵循 suanfa.md 体系最强战法)。
体系最强战法流程(suanfa.md):
1. 日线底背离 → 纳入关注范围
2. 龙抬头出现 → 执行买入操作(实操核心买点)
3. 真龙/主升浪 → 持有仓位+加仓(不是新买入!)
4. 不见主升浪 → 不出场
参数:
signal_status: list[dict] 信号状态列表(来自 signal_detector._check_all_signal_status
indicators: dict 最新技术指标(含 macd.dif, macd.dea 等)
triggered_count: int 触发信号数量
is_holding: bool 当前是否持仓该股票
返回:
tuple: (signal_type, display_text, reason, recommend_rate)
- signal_type: 'buy' | 'sell' | 'watch'
- display_text: '买入' | '卖出' | '加仓' | '持有' | '关注' | '观察' | '观望'
- reason: str 推荐理由
- recommend_rate: int 推荐评分 0-100
使用场景:
- 全景扫描结果推荐列
- 策略建议分档
- 提醒 tab 买卖推荐
- 模拟交易自动买卖决策
"""
if not signal_status:
return ('watch', '观望', '暂无信号数据', 0)
ss = signal_status
sig_map = {}
for s in ss:
sig_map[s.get('type', '')] = s
has_main_wave = sig_map.get('main_rising_wave', {}).get('triggered', False)
has_divergence = sig_map.get('daily_bottom_divergence', {}).get('triggered', False)
has_dragon = sig_map.get('dragon_head', {}).get('triggered', False)
has_real_dragon = sig_map.get('true_dragon', {}).get('triggered', False)
macd = (indicators or {}).get('macd', {})
dif = macd.get('dif', 0)
dea = macd.get('dea', 0)
triggered_signals = [s.get('name', s.get('type', '')) for s in ss if s.get('triggered')]
# ════════════════════════════════════════════
# 持仓逻辑(suanfa.md 步骤3-4
# ════════════════════════════════════════════
if is_holding:
# 卖出条件: MACD死叉 + 无主升浪 → 趋势走弱,不见主升浪则出场
if dif < dea and not has_main_wave:
return ('sell', '卖出',
f"MACD死叉(DIF={dif:.3f}<DEA={dea:.3f})+主升浪消失", 75)
# 主升浪 → 加仓(suanfa.md: 主升浪=持有仓位+加仓)
if has_main_wave:
return ('buy', '加仓', '主升浪信号 → 加速拉升阶段,建议加仓', 90)
# 真龙 → 持有(suanfa.md: 真龙=趋势确认,持有)
if has_real_dragon:
return ('buy', '持有', '真龙信号 → 趋势确立,继续持有', 70)
return ('watch', '观望', '持仓中,等待主升浪信号', 50)
# ════════════════════════════════════════════
# 非持仓逻辑(suanfa.md 步骤1-2
# ════════════════════════════════════════════
# 最佳买入: 底背离+龙抬头(suanfa.md 步骤1→2 的理想组合)
# MACD 死叉时降级,避免与持仓卖出信号矛盾
if has_divergence and has_dragon:
macd_golden = (dif is None or dea is None or dif >= dea)
if macd_golden:
return ('buy', '买入', '日线底背离+龙抬头 → 最佳买入信号', 95)
return ('watch', '关注',
f'底背离+龙抬头但MACD死叉(DIF={dif:.3f}<DEA={dea:.3f}) → 等待MACD金叉确认', 65)
# 核心买入: 龙抬头(suanfa.md: "龙抬头出现 → 执行买入操作")
# 但需要 MACD 配合:如果 MACD 死叉则信号冲突,降级为关注
if has_dragon:
macd_ok = (dif is None or dea is None or dif >= dea) # MACD 金叉或无数据
if has_main_wave and macd_ok:
return ('buy', '买入', '龙抬头+主升浪 → 强势买入信号', 90)
if macd_ok:
return ('buy', '买入', '龙抬头出现 → 短线起爆点,执行买入', 80)
# MACD死叉 + 龙抬头 → 信号冲突,降级为关注
return ('watch', '关注',
f'龙抬头出现但MACD死叉(DIF={dif:.3f}<DEA={dea:.3f}) → 信号冲突,谨慎观望', 55)
# 主升浪(非持仓)→ 不是新买入!suanfa.md: 主升浪=加速段,属于持有/加仓信号
if has_main_wave:
return ('watch', '关注', '主升浪(加速段) → 已过最佳买点,关注回调机会', 75)
# 真龙 → 关注(趋势刚启动,可跟踪)
if has_real_dragon:
return ('watch', '关注', '真龙出现 → 趋势启动,等待龙抬头确认', 65)
# MACD死叉 → 回避(非持仓不能卖出,应为回避/观望)
if dif is not None and dea is not None and dif < dea:
return ('watch', '回避', f"MACD死叉(DIF={dif:.3f}<DEA={dea:.3f}),趋势偏弱", 25)
# 底背离 → 关注(suanfa.md 步骤1: 纳入关注范围)
if has_divergence:
return ('watch', '关注', '日线底背离出现 → 纳入关注,等待龙抬头', 60)
# 其他信号 → 观察
if triggered_count and triggered_count > 0:
sigs = ''.join(triggered_signals[:3])
return ('watch', '观察', f"触发{triggered_count}个信号: {sigs}", 40)
return ('watch', '观望', '无核心信号触发', 0)
# ═══════════════════════════════════════════════
# 6. 获取最新价格
# ═══════════════════════════════════════════════
def get_latest_price(stock_code):
"""
获取股票最新价格(从本地 stock_realtime_price 表)
返回:
float: 最新价格, 失败返回 0
"""
from db import get_db, put_db
conn = get_db()
if not conn:
return 0
try:
cur = conn.cursor()
cur.execute("""
SELECT price FROM stock_realtime_price
WHERE code = %s AND price > 0
""", (stock_code,))
row = cur.fetchone()
if row:
return float(row[0])
except Exception:
pass
finally:
put_db(conn)
return 0
# ═══════════════════════════════════════════════
# 7. 牛股阶段识别(suanfa.md 标准牛股启动信号先后顺序)
# ═══════════════════════════════════════════════
# 标准牛股启动流程(底部→拉升):
# 阶段1 → 日线底背离/短底背离(跌到底部,停止下跌)
# 阶段2 → 龙抬头(资金进场,短线起爆)
# 阶段3 → 真龙(趋势正式确立)
# 阶段4 → ★主升浪(进入加速段,利润兑现最快)
# 阶段5 → 反弹(中途回调后的补涨信号)
# 补充 → 老鼠仓可在底部任意位置提前出现
BULL_STAGES = {
1: {'name': '底部探测', 'icon': '', 'color': '#2196F3',
'desc': '日线底背离/短底背离 → 跌到底部,停止下跌'},
2: {'name': '资金进场', 'icon': '', 'color': '#4CAF50',
'desc': '龙抬头 → 资金进场,短线起爆点(最佳买入时机)'},
3: {'name': '趋势确立', 'icon': '', 'color': '#FF9800',
'desc': '真龙 → 中期趋势正式确立'},
4: {'name': '加速拉升', 'icon': '', 'color': '#F44336',
'desc': '★主升浪 → 进入加速段,利润兑现最快'},
5: {'name': '回调补涨', 'icon': '', 'color': '#9C27B0',
'desc': '反弹 → 中途回调后的补涨信号'},
}
def compute_bull_stage(signal_status):
"""
识别股票在标准牛股启动流程中的阶段。
参数:
signal_status: list[dict] 信号状态列表
返回:
dict: {
'stage': int (0-5, 0=未进入流程),
'stage_name': str,
'stage_icon': str,
'stage_color': str,
'stage_desc': str,
'signals_active': list[str], # 当前活跃的信号名称
'progress': int (0-100), # 牛股流程进度百分比
'next_signal': str, # 下一个期待的信号
'investment_advice': str, # 投资建议
'has_rat_trading': bool, # 是否有老鼠仓(提前埋伏信号)
}
"""
if not signal_status:
return {
'stage': 0, 'stage_name': '观望', 'stage_icon': '',
'stage_color': '#9E9E9E', 'stage_desc': '无信号触发',
'signals_active': [], 'progress': 0,
'next_signal': '等待底背离/短底背离', 'investment_advice': '暂无操作机会',
'has_rat_trading': False,
}
sig_map = {}
for s in signal_status:
sig_map[s.get('type', '')] = s
has_divergence = sig_map.get('daily_bottom_divergence', {}).get('triggered', False)
has_short_div = sig_map.get('short_bottom_divergence', {}).get('triggered', False)
has_dragon = sig_map.get('dragon_head', {}).get('triggered', False)
has_true_dragon = sig_map.get('true_dragon', {}).get('triggered', False)
has_main_wave = sig_map.get('main_rising_wave', {}).get('triggered', False)
has_rebound = sig_map.get('rebound', {}).get('triggered', False)
has_rat = sig_map.get('rat_trading', {}).get('triggered', False)
signals_active = []
if has_divergence:
signals_active.append('日线底背离')
if has_short_div:
signals_active.append('短底背离')
if has_dragon:
signals_active.append('龙抬头')
if has_true_dragon:
signals_active.append('真龙')
if has_main_wave:
signals_active.append('主升浪')
if has_rebound:
signals_active.append('反弹')
if has_rat:
signals_active.append('老鼠仓')
# 确定阶段(按最高阶段判定)
stage = 0
if has_main_wave:
stage = 4
elif has_true_dragon:
stage = 3
elif has_dragon:
stage = 2
elif has_divergence or has_short_div:
stage = 1
elif has_rebound:
stage = 5
elif has_rat:
stage = 1 # 老鼠仓归入底部阶段
if stage == 0:
return {
'stage': 0, 'stage_name': '观望', 'stage_icon': '',
'stage_color': '#9E9E9E', 'stage_desc': '无核心信号触发',
'signals_active': signals_active, 'progress': 0,
'next_signal': '等待底背离/短底背离',
'investment_advice': '暂无操作机会',
'has_rat_trading': has_rat,
}
info = BULL_STAGES[stage]
# 计算流程进度(越靠后越高)
# 加分项:多信号叠加说明流程更完整
base_progress = {1: 20, 2: 45, 3: 65, 4: 85, 5: 50}
progress = base_progress.get(stage, 0)
if stage <= 2 and has_divergence:
progress += 10 # 有底背离做基础更好
if stage >= 2 and has_dragon:
progress += 5
if stage >= 3 and has_true_dragon:
progress += 5
if has_rat:
progress += 5 # 老鼠仓加分
progress = min(progress, 100)
# 下一步信号期待
next_signals = {
1: '等待龙抬头(资金进场信号)',
2: '等待真龙(趋势确认信号)',
3: '等待主升浪(加速拉升信号)',
4: '持有!不见主升浪消失不出场',
5: '等待龙抬头/真龙确认趋势',
}
# 投资建议
advices = {
1: '纳入关注池,等待龙抬头出现后买入',
2: '最佳买入时机!龙抬头=实操核心买点',
3: '趋势已确立,可以追入,建议等回调买入',
4: '已在加速段,持仓者加仓/持有,新入者谨慎追高',
5: '回调中可关注,但需确认不是假反弹',
}
return {
'stage': stage,
'stage_name': info['name'],
'stage_icon': info['icon'],
'stage_color': info['color'],
'stage_desc': info['desc'],
'signals_active': signals_active,
'progress': progress,
'next_signal': next_signals.get(stage, ''),
'investment_advice': advices.get(stage, ''),
'has_rat_trading': has_rat,
}
def find_bull_stocks(scan_rows, holding_codes=None, scores_map=None):
"""
从扫描结果中找出潜在牛股,按阶段分组排序。
参数:
scan_rows: list[dict] 扫描结果列表 (含 code, name, signal_status, indicators, triggered_count)
holding_codes: set 持仓代码集合
scores_map: dict 综合评分映射 {code: {technical_score, external_score, final_score, verdict}}
当提供时,每只股票附加三项得分,并按综合得分排序
返回:
dict: {
'stages': {1: [...], 2: [...], ...}, # 按阶段分组的股票列表
'summary': {1: count, 2: count, ...}, # 各阶段数量统计
'total': int, # 有信号的总数
}
"""
if holding_codes is None:
holding_codes = set()
stages = {1: [], 2: [], 3: [], 4: [], 5: []}
summary = {0: 0, 1: 0, 2: 0, 3: 0, 4: 0, 5: 0}
for row in scan_rows:
signal_status = row.get('signal_status') or []
if not signal_status:
summary[0] += 1
continue
# 判断牛股阶段
bull = compute_bull_stage(signal_status)
stage = bull['stage']
summary[stage] += 1
if stage == 0:
continue
# 计算推荐
is_holding = row.get('code', '') in holding_codes
st, disp, reason, rate = compute_recommend(
signal_status, row.get('indicators'),
row.get('triggered_count'), is_holding,
)
# 附加综合评分(如果提供了 scores_map)
code = row.get('code', '')
technical_score = rate
external_score = 0
final_score = rate
verdict = ''
if scores_map and code in scores_map:
sc = scores_map[code]
technical_score = sc.get('technical_score', rate)
external_score = sc.get('external_score', 0)
final_score = sc.get('final_score', rate)
verdict = sc.get('verdict', '')
item = {
'code': code,
'name': row.get('name', ''),
'stage': stage,
'stage_name': bull['stage_name'],
'stage_icon': bull['stage_icon'],
'stage_color': bull['stage_color'],
'signals_active': bull['signals_active'],
'progress': bull['progress'],
'next_signal': bull['next_signal'],
'investment_advice': bull['investment_advice'],
'has_rat_trading': bull['has_rat_trading'],
'recommend_type': st,
'recommend_text': disp,
'recommend_reason': reason,
'recommend_rate': final_score,
'is_holding': is_holding,
'triggered_count': row.get('triggered_count', 0),
'technical_score': technical_score,
'external_score': external_score,
'final_score': final_score,
'verdict': verdict,
}
stages[stage].append(item)
# 每个阶段内排序:有综合评分时按综合得分→技术得分→进度,否则按推荐评分→进度
if scores_map:
for stage_num in stages:
stages[stage_num].sort(
key=lambda x: (-x.get('final_score', 0), -x.get('technical_score', 0), -x['progress'])
)
else:
for stage_num in stages:
stages[stage_num].sort(key=lambda x: (-x['recommend_rate'], -x['progress']))
total = sum(len(v) for v in stages.values())
return {
'stages': stages,
'summary': summary,
'total': total,
'stage_info': BULL_STAGES,
}
# ═══════════════════════════════════════════════
# 9. 单股深度分析(价格位置、压力支撑、量价、空间估算)
# ═══════════════════════════════════════════════
def _generate_plain_summary(price, change_pct, ma_trend, position, supports,
resistances, vol_ratio, vol_trend, patterns,
space, score, verdict, reasons):
"""根据技术分析结果生成通俗易懂的中文解说"""
parts = []
# 1. 当前走势概况
if change_pct > 3:
trend_desc = f'今天涨了{change_pct:.1f}%,涨势比较猛'
elif change_pct > 0:
trend_desc = f'今天小涨{change_pct:.1f}%'
elif change_pct > -3:
trend_desc = f'今天小跌{abs(change_pct):.1f}%'
else:
trend_desc = f'今天跌了{abs(change_pct):.1f}%,跌幅较大'
if ma_trend == 'bullish':
trend_desc += ',均线呈多头排列,说明中短期整体向上'
elif ma_trend == 'bearish':
trend_desc += ',均线呈空头排列,中短期趋势偏弱'
else:
trend_desc += ',均线交叉纠缠,短期方向还不太明确'
parts.append(trend_desc + '')
# 2. 价格位置(用大白话)
pos_20 = position.get('d20', {})
pct_20 = pos_20.get('range_pct', 50)
if pct_20 > 80:
parts.append(f'当前股价处于近20天的高位区间({pct_20:.0f}%位置),已经涨了不少,追高要小心。')
elif pct_20 > 50:
parts.append(f'股价在近20天的中高位置({pct_20:.0f}%),还有一定上涨空间。')
elif pct_20 > 20:
parts.append(f'股价在近20天的中低位置({pct_20:.0f}%),相对安全。')
else:
parts.append(f'股价处于近20天的低位区间({pct_20:.0f}%),可能存在反弹机会。')
# 3. 上方压力和下方支撑
if resistances:
nearest_r = resistances[0]
r_gap = round((nearest_r['level'] - price) / price * 100, 1) if price > 0 else 0
if r_gap > 0:
parts.append(f'往上最近的压力位在{nearest_r["level"]:.2f}元({nearest_r["name"]}),距离约{r_gap:.1f}%。')
if supports:
nearest_s = supports[0]
s_gap = round((price - nearest_s['level']) / price * 100, 1) if price > 0 else 0
if s_gap > 0:
parts.append(f'往下最近的支撑位在{nearest_s["level"]:.2f}元({nearest_s["name"]}),有{s_gap:.1f}%的安全垫。')
# 4. 成交量情况
if vol_ratio >= 2:
parts.append(f'成交量明显放大(量比{vol_ratio:.1f}倍),市场关注度很高,要留意是主力进场还是出货。')
elif vol_ratio >= 1.3:
parts.append(f'成交量温和放大(量比{vol_ratio:.1f}倍),有资金在活跃参与。')
elif vol_ratio < 0.6:
parts.append(f'成交量萎缩(量比{vol_ratio:.1f}倍),市场比较冷清,短期可能震荡。')
else:
parts.append(f'成交量正常(量比{vol_ratio:.1f}倍)。')
# 5. 形态识别
if patterns:
pattern_names = [p['name'] for p in patterns]
bullish_p = [p['name'] for p in patterns if p.get('bullish') is True]
bearish_p = [p['name'] for p in patterns if p.get('bullish') is False]
if bullish_p:
parts.append(f'发现看涨信号:{"".join(bullish_p)},这是积极的技术形态。')
if bearish_p:
parts.append(f'注意看跌信号:{"".join(bearish_p)},需要警惕。')
# 6. 综合建议(大白话)
action_tip = ''
if score >= 75:
action_tip = '综合来看比较乐观,可以考虑逢低关注或适量参与,但注意控制仓位。'
elif score >= 60:
action_tip = '整体偏积极,可以少量关注,等回调到支撑位附近再考虑。'
elif score >= 45:
action_tip = '目前多空力量比较均衡,建议观望为主,等方向更明确再做决定。'
elif score >= 30:
action_tip = '目前偏弱势,不建议急于买入。如果持有,可以在反弹时适当减仓。'
else:
action_tip = '当前走势比较弱,建议回避。已经持有的可以考虑止损或等待反弹减仓。'
# 7. 空间估算
rr = space.get('risk_reward', 0)
if rr and rr > 0:
if rr >= 2:
parts.append(f'从空间来看,潜在收益是风险的{rr:.1f}倍,性价比不错。')
elif rr >= 1:
parts.append(f'收益风险比{rr:.1f}:1,性价比一般。')
else:
parts.append(f'收益风险比仅{rr:.1f}:1,下行风险大于上涨空间,不太划算。')
summary_text = ''.join(parts)
return {
'text': summary_text,
'action_tip': action_tip,
'confidence': '' if score >= 70 or score <= 30 else '',
}
def compute_deep_analysis(df, signal_result=None, realtime_info=None):
"""
对单只股票进行深度分析,返回结构化的分析报告。
参数:
df: DataFrame (含技术指标的K线数据)
signal_result: dict (detect_all_signals 返回的结果,可选)
realtime_info: dict (stock_realtime_price 行数据,可选)
返回:
dict: 完整的深度分析报告
"""
import numpy as np
if df is None or len(df) < 30:
return {'error': 'K线数据不足(需要至少30天)'}
last = df.iloc[-1]
cl = float(last['close'])
n = len(df)
# ---- 1. 均线系统 ----
ma_data = {}
for period in [5, 10, 20, 60]:
col = f'ma{period}'
if col in df.columns and n >= period:
ma_data[f'ma{period}'] = round(float(df[col].iloc[-1]), 2)
ma_list = sorted(ma_data.items(), key=lambda x: x[1], reverse=True)
ma_trend = 'bullish' if all(
ma_data.get(f'ma{a}', 0) >= ma_data.get(f'ma{b}', 0)
for a, b in [(5, 10), (10, 20)]
) else 'bearish' if all(
ma_data.get(f'ma{a}', 0) <= ma_data.get(f'ma{b}', 0)
for a, b in [(5, 10), (10, 20)]
) else 'mixed'
ma_trend_label = {'bullish': '多头排列', 'bearish': '空头排列', 'mixed': '交叉整理'}
# ---- 2. 价格位置分析 ----
position = {}
for days in [20, 60, 120]:
subset = df.tail(days) if n >= days else df
h = float(subset['high'].max())
l = float(subset['low'].min())
rng = h - l
pct = round((cl - l) / rng * 100, 0) if rng > 0 else 50
position[f'd{days}'] = {
'high': round(h, 2), 'low': round(l, 2),
'range_pct': pct,
'up_space': round((h / cl - 1) * 100, 1),
'down_risk': round((1 - l / cl) * 100, 1),
}
# ---- 3. 支撑与压力位 ----
supports = []
resistances = []
for name, val in ma_data.items():
if val < cl:
supports.append({'level': val, 'type': 'ma', 'name': name.upper()})
elif val > cl:
resistances.append({'level': val, 'type': 'ma', 'name': name.upper()})
for days_key in ['d20', 'd60', 'd120']:
p = position.get(days_key, {})
label = days_key.replace('d', '') + ''
if p.get('low', 0) < cl:
supports.append({'level': p['low'], 'type': 'low', 'name': f'{label}低点'})
if p.get('high', 0) > cl:
resistances.append({'level': p['high'], 'type': 'high', 'name': f'{label}高点'})
supports.sort(key=lambda x: x['level'], reverse=True)
resistances.sort(key=lambda x: x['level'])
# ---- 4. 成交量分析 ----
vol = float(last['volume'])
vol_5 = float(df['volume'].tail(5).mean()) if n >= 5 else vol
vol_20 = float(df['volume'].tail(20).mean()) if n >= 20 else vol
vol_ratio = round(vol / vol_20, 1) if vol_20 > 0 else 1.0
vol_trend = '缩量' if vol_ratio < 0.7 else '平量' if vol_ratio < 1.3 else '温和放量' if vol_ratio < 2.0 else '大幅放量'
# ---- 5. 形态识别(增强版) ----
patterns = []
closes_10 = [float(x) for x in df['close'].tail(10)]
if n >= 10:
std_10 = np.std(closes_10)
mean_10 = np.mean(closes_10)
cv_10 = std_10 / mean_10 if mean_10 > 0 else 0
if cv_10 < 0.015 and cl > max(closes_10[:-1]):
patterns.append({'name': '平台突破', 'bullish': True,
'desc': f'近10日波动率仅{cv_10*100:.1f}%,今日突破平台'})
elif cv_10 < 0.015:
patterns.append({'name': '窄幅整理', 'bullish': None,
'desc': f'近10日波动率{cv_10*100:.1f}%,蓄势待变'})
if n >= 20:
h20 = float(df.tail(20)['high'].max())
if cl >= h20 * 0.99:
patterns.append({'name': '创20日新高', 'bullish': True,
'desc': f'触及20日高点{h20:.2f}'})
# 双底形态:近30日内两个低点价格接近(差异<3%),且当前价格高于两低点之间的高点
if n >= 30:
lows_30 = [float(x) for x in df['low'].tail(30)]
# 找最低点和次低点
min_idx = int(np.argmin(lows_30))
min_val = lows_30[min_idx]
# 在最低点之前找次低点
if min_idx > 5:
before_lows = lows_30[:min_idx]
second_min_idx = int(np.argmin(before_lows))
second_min_val = before_lows[second_min_idx]
if abs(min_val - second_min_val) / min_val < 0.03:
# 两低点之间的高点
between_high = max(lows_30[second_min_idx:min_idx])
if cl > between_high:
patterns.append({'name': '双底突破', 'bullish': True,
'desc': f'双底形态(低点{min_val:.2f}{second_min_val:.2f}),已突破颈线{between_high:.2f}'})
# 量价齐升:近5日成交量递增且价格递增
if n >= 5:
vols_5 = [float(x) for x in df['volume'].tail(5)]
closes_5 = [float(x) for x in df['close'].tail(5)]
if all(vols_5[i] <= vols_5[i+1] for i in range(len(vols_5)-1)) and \
all(closes_5[i] <= closes_5[i+1] for i in range(len(closes_5)-1)):
patterns.append({'name': '量价齐升', 'bullish': True,
'desc': '近5日成交量与价格同步递增,强势特征'})
# 均线粘合后发散:MA5/10/20 三线粘合后开始发散
if n >= 20:
ma5_val = ma_data.get('ma5', 0)
ma10_val = ma_data.get('ma10', 0)
ma20_val = ma_data.get('ma20', 0)
if ma5_val and ma10_val and ma20_val:
ma_spread = max(ma5_val, ma10_val, ma20_val) - min(ma5_val, ma10_val, ma20_val)
ma_pct = ma_spread / cl * 100
if ma_pct < 1.0 and ma5_val > ma10_val > ma20_val:
patterns.append({'name': '均线粘合发散', 'bullish': True,
'desc': f'MA5/10/20粘合(离散{ma_pct:.1f}%)后多头排列'})
# 涨跌幅计算:如果最后一条是今天(可能未收盘),用前一日收盘价计算
from datetime import date
last_date_str = str(df['date'].values[-1])[:10]
today_str = date.today().isoformat()
if last_date_str == today_str and n >= 3:
# 今天未收盘,用倒数第二根K线的收盘价对比倒数第三根
change_today = round((cl / float(df.iloc[-2]['close']) - 1) * 100, 2)
else:
change_today = round((cl / float(df.iloc[-2]['close']) - 1) * 100, 2) if n >= 2 else 0
if change_today >= 5:
patterns.append({'name': '大阳线', 'bullish': True,
'desc': f'涨幅{change_today:.1f}%'})
elif change_today <= -5:
patterns.append({'name': '大阴线', 'bullish': False,
'desc': f'跌幅{change_today:.1f}%'})
# ---- 6. 空间估算 ----
first_resist = resistances[0] if resistances else None
first_support = supports[0] if supports else None
space = {
'nearest_resist': first_resist,
'nearest_support': first_support,
'risk_reward': None,
}
if first_resist and first_support:
upside = first_resist['level'] - cl
downside = cl - first_support['level']
space['risk_reward'] = round(upside / downside, 1) if downside > 0 else 99
# ---- 7. 综合评估 ----
score = 50
reasons = []
if ma_trend == 'bullish':
score += 10
reasons.append('均线多头排列(+10)')
elif ma_trend == 'bearish':
score -= 10
reasons.append('均线空头排列(-10)')
if vol_ratio >= 1.3:
score += 5
reasons.append(f'放量{vol_ratio}倍(+5)')
elif vol_ratio < 0.6:
score -= 3
reasons.append(f'缩量{vol_ratio}倍(-3)')
any_breakout = any(p['name'] == '平台突破' for p in patterns)
if any_breakout:
score += 10
reasons.append('平台突破(+10)')
any_new_high = any(p['name'] == '创20日新高' for p in patterns)
if any_new_high:
score += 5
reasons.append('创20日新高(+5)')
any_double_bottom = any(p['name'] == '双底突破' for p in patterns)
if any_double_bottom:
score += 10
reasons.append('双底突破(+10)')
any_vol_price_rise = any(p['name'] == '量价齐升' for p in patterns)
if any_vol_price_rise:
score += 8
reasons.append('量价齐升(+8)')
any_ma_converge = any(p['name'] == '均线粘合发散' for p in patterns)
if any_ma_converge:
score += 7
reasons.append('均线粘合发散(+7)')
pos_120 = position.get('d120', {}).get('range_pct', 50)
if pos_120 < 30:
score += 5
reasons.append(f'120日位置偏低{pos_120}%(+5)')
elif pos_120 > 80:
score -= 5
reasons.append(f'120日位置偏高{pos_120}%(-5)')
# 20日位置也纳入评分
pos_20 = position.get('d20', {}).get('range_pct', 50)
if pos_20 < 25:
score += 3
reasons.append(f'20日位置偏低{pos_20}%(+3)')
elif pos_20 > 85:
score -= 3
reasons.append(f'20日位置偏高{pos_20}%(-3)')
if signal_result:
sig_count = signal_result.get('signal_summary', {}).get('total_signals', 0)
if sig_count >= 3:
score += 15
reasons.append(f'{sig_count}信号共振(+15)')
elif sig_count >= 2:
score += 10
reasons.append(f'{sig_count}信号叠加(+10)')
elif sig_count >= 1:
score += 5
reasons.append(f'{sig_count}个信号(+5)')
if space.get('risk_reward') and space['risk_reward'] >= 2:
score += 5
reasons.append(f'风险收益比{space["risk_reward"]}:1(+5)')
elif space.get('risk_reward') and space['risk_reward'] < 0.8:
score -= 5
reasons.append(f'风险收益比{space["risk_reward"]}:1(-5)')
score = max(0, min(100, score))
verdict = '强烈看多' if score >= 80 else '看多' if score >= 65 else '中性偏多' if score >= 50 else '中性偏空' if score >= 35 else '看空'
ai_summary = _generate_plain_summary(
cl, change_today, ma_trend, position, supports, resistances,
vol_ratio, vol_trend, patterns, space, score, verdict, reasons
)
return {
'price': cl,
'change_pct': change_today,
'ma': ma_data,
'ma_trend': ma_trend,
'ma_trend_label': ma_trend_label[ma_trend],
'position': position,
'supports': supports[:5],
'resistances': resistances[:5],
'volume': {
'today': vol,
'avg_5': round(vol_5),
'avg_20': round(vol_20),
'ratio': vol_ratio,
'trend': vol_trend,
},
'patterns': patterns,
'space': space,
'deep_score': score,
'verdict': verdict,
'score_reasons': reasons,
'ai_summary': ai_summary,
'kline_days': n,
}