hahaju/aiagents-stock
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

增加智瞰龙虎板块

Browse Source
This commit is contained in:
oficcejo
2025-10-17 19:26:13 +08:00
parent b127ffe60d
commit 18b63dd9c9
5 changed files with 1133 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
longhubang_scoring.py
+582
View File
@@ -0,0 +1,582 @@
"""
智瞰龙虎AI智能评分模块
对龙虎榜上榜股票进行综合评分排名
"""
import pandas as pd
from typing import Dict, List
from collections import Counter
class LonghubangScoring:
"""龙虎榜股票智能评分系统"""
def __init__(self):
"""初始化评分系统"""
# 顶级游资名单（根据市场知名度和历史战绩）
self.top_youzi = [
'赵老哥', '章盟主', '92科比', '瑞鹤仙', '小鳄鱼',
'养家心法', '欢乐海岸', '古北路', '成都系', '佛山系',
'方新侠', '乔帮主', '淮海路', '东方财富',
'国信深圳', '华泰深圳', '中信杭州', '招商深圳'
]
# 知名游资（次一级）
self.famous_youzi = [
'深股通', '沪股通', '北向资金',
'中金公司', '中信证券', '国泰君安', '海通证券',
'广发证券', '华泰证券', '招商证券'
]
# 机构关键词
self.institution_keywords = [
'机构专用', '机构', '基金', '保险', '社保',
'QFII', 'RQFII', '券商', '信托'
]
print("[智瞰龙虎] 评分系统初始化完成")
def calculate_stock_score(self, stock_data: List[Dict]) -> float:
"""
计算单个股票的综合评分
Args:
stock_data: 该股票的所有龙虎榜记录
Returns:
综合评分 (0-100分)
"""
if not stock_data:
return 0.0
# 1. 买入资金含金量评分 (0-30分)
capital_quality_score = self._calculate_capital_quality(stock_data)
# 2. 净买入额评分 (0-25分)
net_inflow_score = self._calculate_net_inflow_score(stock_data)
# 3. 卖出压力评分 (0-20分)
sell_pressure_score = self._calculate_sell_pressure_score(stock_data)
# 4. 机构共振评分 (0-15分)
institution_score = self._calculate_institution_score(stock_data)
# 5. 其他加分项 (0-10分)
bonus_score = self._calculate_bonus_score(stock_data)
# 综合评分
total_score = (
capital_quality_score +
net_inflow_score +
sell_pressure_score +
institution_score +
bonus_score
)
return round(total_score, 1)
def _calculate_capital_quality(self, stock_data: List[Dict]) -> float:
"""
计算买入资金含金量评分 (0-30分)
顶级游资加分多，普通游资加分少
"""
score = 0.0
max_score = 30.0
buyers = []
for record in stock_data:
buy_amount = record.get('买入金额', 0) or record.get('mrje', 0)
# 确保转换为数值类型
try:
buy_amount = float(buy_amount) if buy_amount else 0
except (ValueError, TypeError):
buy_amount = 0
if buy_amount > 0:
youzi_name = record.get('游资名称', '') or record.get('yzmc', '')
yingye_bu = record.get('营业部', '') or record.get('yyb', '')
buyers.append({
'name': youzi_name,
'yingye_bu': yingye_bu,
'amount': float(buy_amount)
})
if not buyers:
return 0.0
# 顶级游资：每个加8-10分
top_youzi_count = 0
for buyer in buyers:
for top in self.top_youzi:
if top in buyer['name'] or top in buyer['yingye_bu']:
top_youzi_count += 1
score += 10.0
break
# 知名游资：每个加4-6分
famous_youzi_count = 0
for buyer in buyers:
is_top = any(top in buyer['name'] or top in buyer['yingye_bu']
for top in self.top_youzi)
if not is_top:
for famous in self.famous_youzi:
if famous in buyer['name'] or famous in buyer['yingye_bu']:
famous_youzi_count += 1
score += 5.0
break
# 普通游资：每个加1-2分
ordinary_count = len(buyers) - top_youzi_count - famous_youzi_count
score += ordinary_count * 1.5
# 限制最高分
return min(score, max_score)
def _calculate_net_inflow_score(self, stock_data: List[Dict]) -> float:
"""
计算净买入额评分 (0-25分)
真金白银越多分数越高
"""
max_score = 25.0
# 计算总净流入
total_net_inflow = 0.0
for record in stock_data:
net_inflow = record.get('净流入金额', 0) or record.get('jlrje', 0)
try:
net_inflow = float(net_inflow) if net_inflow else 0
total_net_inflow += net_inflow
except (ValueError, TypeError):
pass
if total_net_inflow <= 0:
return 0.0
# 净流入分段评分
# 1000万以下：0-10分
# 1000-5000万：10-18分
# 5000万-1亿：18-22分
# 1亿以上：22-25分
net_inflow_wan = total_net_inflow / 10000 # 转换为万元
if net_inflow_wan < 1000:
score = (net_inflow_wan / 1000) * 10
elif net_inflow_wan < 5000:
score = 10 + ((net_inflow_wan - 1000) / 4000) * 8
elif net_inflow_wan < 10000:
score = 18 + ((net_inflow_wan - 5000) / 5000) * 4
else:
score = 22 + min((net_inflow_wan - 10000) / 10000, 1) * 3
return min(score, max_score)
def _calculate_sell_pressure_score(self, stock_data: List[Dict]) -> float:
"""
计算卖出压力评分 (0-20分)
卖出压力越小分数越高
"""
max_score = 20.0
total_buy = 0.0
total_sell = 0.0
for record in stock_data:
buy_amount = record.get('买入金额', 0) or record.get('mrje', 0)
sell_amount = record.get('卖出金额', 0) or record.get('mcje', 0)
try:
buy_amount = float(buy_amount) if buy_amount else 0
total_buy += buy_amount
except (ValueError, TypeError):
pass
try:
sell_amount = float(sell_amount) if sell_amount else 0
total_sell += sell_amount
except (ValueError, TypeError):
pass
if total_buy == 0:
return 0.0
# 计算卖出比例
sell_ratio = total_sell / total_buy if total_buy > 0 else 1.0
# 卖出压力评分
# 卖出比例0-10%：20分
# 卖出比例10-30%：15-20分
# 卖出比例30-50%：10-15分
# 卖出比例50-80%：5-10分
# 卖出比例80%以上：0-5分
if sell_ratio < 0.1:
score = 20.0
elif sell_ratio < 0.3:
score = 20.0 - (sell_ratio - 0.1) / 0.2 * 5
elif sell_ratio < 0.5:
score = 15.0 - (sell_ratio - 0.3) / 0.2 * 5
elif sell_ratio < 0.8:
score = 10.0 - (sell_ratio - 0.5) / 0.3 * 5
else:
score = 5.0 - min(sell_ratio - 0.8, 0.2) / 0.2 * 5
return max(0, min(score, max_score))
def _calculate_institution_score(self, stock_data: List[Dict]) -> float:
"""
计算机构共振评分 (0-15分)
机构+游资共振最高分
"""
max_score = 15.0
has_institution = False
has_youzi = False
institution_count = 0
youzi_count = 0
for record in stock_data:
buy_amount = record.get('买入金额', 0) or record.get('mrje', 0)
try:
buy_amount = float(buy_amount) if buy_amount else 0
except (ValueError, TypeError):
buy_amount = 0
if buy_amount <= 0:
continue
youzi_name = record.get('游资名称', '') or record.get('yzmc', '')
yingye_bu = record.get('营业部', '') or record.get('yyb', '')
# 检查是否是机构
if any(keyword in youzi_name or keyword in yingye_bu
for keyword in self.institution_keywords):
has_institution = True
institution_count += 1
else:
has_youzi = True
youzi_count += 1
# 评分逻辑
if has_institution and has_youzi:
# 机构+游资共振：最高分
score = 15.0
elif has_institution:
# 仅机构：8-12分
score = min(8 + institution_count * 2, 12)
elif has_youzi:
# 仅游资：5-10分
score = min(5 + youzi_count * 1, 10)
else:
score = 0.0
return min(score, max_score)
def _calculate_bonus_score(self, stock_data: List[Dict]) -> float:
"""
计算其他加分项 (0-10分)
买卖比例、主力集中度、热门概念等
"""
max_score = 10.0
score = 0.0
if not stock_data:
return 0.0
# 1. 主力集中度加分 (0-3分)
# 如果资金集中在少数几个席位，说明主力信心强
seat_count = len(stock_data)
if seat_count == 1:
score += 3.0
elif seat_count == 2:
score += 2.5
elif seat_count == 3:
score += 2.0
elif seat_count <= 5:
score += 1.5
else:
score += 1.0
# 2. 热门概念加分 (0-3分)
all_concepts = []
for record in stock_data:
concepts = record.get('概念', '') or record.get('gl', '')
if concepts:
all_concepts.extend([c.strip() for c in str(concepts).split(',')])
hot_keywords = [
'人工智能', 'AI', 'ChatGPT', '算力', '新能源', '芯片', '半导体',
'军工', '医药', '消费', '5G', '新材料', '量子', '光伏',
'储能', '锂电池', '汽车', '游戏', '传媒', '元宇宙'
]
concept_score = 0
for concept in all_concepts:
if any(keyword in concept for keyword in hot_keywords):
concept_score += 0.3
score += min(concept_score, 3.0)
# 3. 连续上榜加分 (0-2分)
# 这里简化处理，如果有多条记录可能表示连续上榜
if len(stock_data) >= 3:
score += 2.0
elif len(stock_data) == 2:
score += 1.0
# 4. 买卖比例优秀加分 (0-2分)
total_buy = 0.0
total_sell = 0.0
for r in stock_data:
try:
buy = float(r.get('买入金额', 0) or r.get('mrje', 0) or 0)
total_buy += buy
except (ValueError, TypeError):
pass
try:
sell = float(r.get('卖出金额', 0) or r.get('mcje', 0) or 0)
total_sell += sell
except (ValueError, TypeError):
pass
if total_buy > 0:
buy_sell_ratio = total_buy / (total_sell + 1)
if buy_sell_ratio >= 10:
score += 2.0
elif buy_sell_ratio >= 5:
score += 1.5
elif buy_sell_ratio >= 3:
score += 1.0
return min(score, max_score)
def score_all_stocks(self, data_list: List[Dict]) -> pd.DataFrame:
"""
对所有上榜股票进行评分排名
Args:
data_list: 龙虎榜数据列表
Returns:
评分排名DataFrame
"""
if not data_list:
return pd.DataFrame()
# 按股票代码分组
stocks_dict = {}
for record in data_list:
code = record.get('股票代码') or record.get('gpdm')
name = record.get('股票名称') or record.get('gpmc')
if not code:
continue
if code not in stocks_dict:
stocks_dict[code] = {
'code': code,
'name': name,
'records': []
}
stocks_dict[code]['records'].append(record)
# 计算每只股票的评分
results = []
for code, stock_info in stocks_dict.items():
records = stock_info['records']
# 计算各维度评分
capital_quality = self._calculate_capital_quality(records)
net_inflow = self._calculate_net_inflow_score(records)
sell_pressure = self._calculate_sell_pressure_score(records)
institution = self._calculate_institution_score(records)
bonus = self._calculate_bonus_score(records)
total_score = capital_quality + net_inflow + sell_pressure + institution + bonus
# 计算实际数据（安全转换）
total_buy = 0.0
total_sell = 0.0
total_net = 0.0
for r in records:
try:
buy = float(r.get('买入金额', 0) or r.get('mrje', 0) or 0)
total_buy += buy
except (ValueError, TypeError):
pass
try:
sell = float(r.get('卖出金额', 0) or r.get('mcje', 0) or 0)
total_sell += sell
except (ValueError, TypeError):
pass
try:
net = float(r.get('净流入金额', 0) or r.get('jlrje', 0) or 0)
total_net += net
except (ValueError, TypeError):
pass
# 统计买入席位数（安全比较）
buy_seats = 0
for r in records:
try:
buy = float(r.get('买入金额', 0) or r.get('mrje', 0) or 0)
if buy > 0:
buy_seats += 1
except (ValueError, TypeError):
pass
# 统计机构数量
institution_count = sum(1 for r in records
if any(kw in (r.get('游资名称', '') or r.get('yzmc', '')) or
kw in (r.get('营业部', '') or r.get('yyb', ''))
for kw in self.institution_keywords))
# 获取概念
concepts_list = []
for r in records:
concepts = r.get('概念', '') or r.get('gl', '')
if concepts:
concepts_list.extend([c.strip() for c in str(concepts).split(',')])
top_concepts = Counter(concepts_list).most_common(3)
concept_str = ','.join([c[0] for c in top_concepts]) if top_concepts else ''
# 判断机构参与
has_institution = institution_count > 0
results.append({
'排名': 0, # 稍后填充
'股票名称': stock_info['name'],
'股票代码': code,
'综合评分': round(total_score, 1),
'资金含金量': round(capital_quality, 0),
'净买入额': round(net_inflow, 0),
'卖出压力': round(sell_pressure, 0),
'机构共振': round(institution, 0),
'加分项': round(bonus, 0),
'顶级游资': self._count_top_youzi(records),
'买方数': buy_seats,
'机构参与': '✅' if has_institution else '❌',
'净流入': round(total_net, 2)
})
# 转换为DataFrame并排序
df = pd.DataFrame(results)
if df.empty:
return df
df = df.sort_values('综合评分', ascending=False).reset_index(drop=True)
df['排名'] = range(1, len(df) + 1)
# 添加奖牌
if len(df) >= 1:
df.loc[0, '排名'] = '🥇 1'
if len(df) >= 2:
df.loc[1, '排名'] = '🥈 2'
if len(df) >= 3:
df.loc[2, '排名'] = '🥉 3'
return df
def _count_top_youzi(self, records: List[Dict]) -> int:
"""统计顶级游资数量"""
count = 0
for record in records:
buy_amount = record.get('买入金额', 0) or record.get('mrje', 0)
try:
buy_amount = float(buy_amount) if buy_amount else 0
except (ValueError, TypeError):
buy_amount = 0
if buy_amount <= 0:
continue
youzi_name = record.get('游资名称', '') or record.get('yzmc', '')
yingye_bu = record.get('营业部', '') or record.get('yyb', '')
if any(top in youzi_name or top in yingye_bu for top in self.top_youzi):
count += 1
return count
def get_score_explanation(self) -> str:
"""获取评分维度说明"""
explanation = """
【AI智能评分维度说明】
📊 总分100分，由5个维度组成：
1️⃣ 买入资金含金量 (0-30分)
- 顶级游资（赵老哥、章盟主等）：每个+10分
- 知名游资（深股通、中信等）：每个+5分
- 普通游资：每个+1.5分
2️⃣ 净买入额评分 (0-25分)
- 净流入1000万以下：0-10分
- 净流入1000-5000万：10-18分
- 净流入5000万-1亿：18-22分
- 净流入1亿以上：22-25分
3️⃣ 卖出压力评分 (0-20分)
- 卖出比例0-10%：20分（压力极小）
- 卖出比例10-30%：15-20分（压力较小）
- 卖出比例30-50%：10-15分（压力中等）
- 卖出比例50-80%：5-10分（压力较大）
- 卖出比例80%以上：0-5分（压力极大）
4️⃣ 机构共振评分 (0-15分)
- 机构+游资共振：15分（最强信号）
- 仅机构买入：8-12分
- 仅游资买入：5-10分
5️⃣ 其他加分项 (0-10分)
- 主力集中度：席位越少越集中，+1-3分
- 热门概念：AI、新能源、芯片等，+0-3分
- 连续上榜：连续多日上榜，+0-2分
- 买卖比例优秀：买入远大于卖出，+0-2分
💡 评分越高，表示该股票受到资金青睐程度越高，
但仍需结合市场环境、技术面等因素综合判断！
"""
return explanation
# 测试函数
if __name__ == "__main__":
print("=" * 60)
print("测试智瞰龙虎评分系统")
print("=" * 60)
# 创建测试数据
test_data = [
{
'股票代码': '001337',
'股票名称': '四川黄金',
'游资名称': '92科比',
'营业部': '兴业证券股份有限公司南京天元东路证券营业部',
'买入金额': 14470401,
'卖出金额': 15080,
'净流入金额': 14455321,
'概念': '贵金属,黄金概念,次新股'
},
{
'股票代码': '001337',
'股票名称': '四川黄金',
'游资名称': '赵老哥',
'营业部': '某证券公司',
'买入金额': 10000000,
'卖出金额': 0,
'净流入金额': 10000000,
'概念': '贵金属,黄金概念'
}
]
scoring = LonghubangScoring()
# 测试评分
df_result = scoring.score_all_stocks(test_data)
print("\n评分结果：")
print(df_result)
print("\n" + scoring.get_score_explanation())