diff --git a/README.md b/README.md index 64e34ab..ba9678c 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,26 @@ - 初心：在股市摸爬滚打多年，自学自编各种指标，花冤柉钱学习了各种战法各种策略，也曾入各种小班，总是赚少赔多，逐渐失去在股市玩的信心。自从去年deepseek上市，一直探索用ai辅助分析，且近日受tradingagents项目启发（感谢原作），多agent结合跟踪主力资金战法（某指每年收费6000rmb），用各种ai辅助编程，拼凑了这么个小程序，根据软件提供的辅助信息，实盘测试盈率还是挺高的，并且逐步形成了自己的交易系统，近一个月来，账户也慢慢在扰亏为盈。开源此软件的目的，就是为了使像我一样的小散，不再迷范。也许这个软件不能让你发大财，但是他能给你足够的信心。最后提醒：股市有风险，入市需谨慎！ +## ⭐ 1214更新 - 小市值策略 📊 + +新增“小市值策略”选股板块，专注小盘高成长股票： + +**核心功能：** +- ✅ **精准筛选** - 总市值≤50亿 + 营收增长率≥10% + 净利润增长率≥100% + 沪深A股 +- ✅ **小盘优先** - 按总市值由小到大排序，优选最小盘股票 +- ✅ **策略监控** - 复用低价擒牛监控框架，自动跟踪持仓 +- ✅ **双重信号** - 持股5天 + MA5下穿MA20智能提醒 +- ✅ **仓位管理** - 资金10万、个股最大3成、最多4股 +- ✅ **铉钉通知** - 选股结果和卖出信号自动通知 + +**使用流程：** +1. 在选股板块中点击“📊 小市值策略” +2. 执行选股，查看筛选结果 +3. 点击“➕ 加入策略监控”加入监控 +4. 在策略监控面板统一管理 + +--- + ## ⭐ 1212更新 - 低价擒牛策略监控 🐂 新增“低价擒牛”选股板块，自动监控持仓并智能提醒卖出： diff --git a/app.py b/app.py index aeaac58..f21e56c 100644 --- a/app.py +++ b/app.py @@ -314,7 +314,14 @@ def main(): if st.button("🐂 低价擒牛", width='stretch', key="nav_low_price_bull", help="低价高成长股票筛选策略"): st.session_state.show_low_price_bull = True for key in ['show_history', 'show_monitor', 'show_config', 'show_sector_strategy', - 'show_longhubang', 'show_portfolio', 'show_main_force']: + 'show_longhubang', 'show_portfolio', 'show_main_force', 'show_small_cap']: + if key in st.session_state: + del st.session_state[key] + + if st.button("📊 小市值策略", width='stretch', key="nav_small_cap", help="小盘高成长股票筛选策略"): + st.session_state.show_small_cap = True + for key in ['show_history', 'show_monitor', 'show_config', 'show_sector_strategy', + 'show_longhubang', 'show_portfolio', 'show_main_force', 'show_low_price_bull']: if key in st.session_state: del st.session_state[key] @@ -473,6 +480,12 @@ def main(): from low_price_bull_ui import display_low_price_bull display_low_price_bull() return + + # 检查是否显示小市值策略 + if 'show_small_cap' in st.session_state and st.session_state.show_small_cap: + from small_cap_ui import display_small_cap + display_small_cap() + return # 检查是否显示智策板块 if 'show_sector_strategy' in st.session_state and st.session_state.show_sector_strategy: diff --git a/docs/小市值策略功能说明.md b/docs/小市值策略功能说明.md new file mode 100644 index 0000000..ff4acee --- /dev/null +++ b/docs/小市值策略功能说明.md @@ -0,0 +1,286 @@ +# 小市值策略功能说明 + +## 📊 功能概述 + +"小市值策略"是专注于小盘高成长股票的选股策略，通过严格的财务指标筛选，挖掘市值小、成长性强的优质股票。 + +## 🎯 筛选条件 + +### 核心指标 + +| 指标 | 条件 | 说明 | +|------|------|------| +| 总市值 | ≤ 50亿 | 小盘股票，成长空间大 | +| 营收增长率 | ≥ 10% | 营业收入持续增长 | +| 净利润增长率 | ≥ 100% | 利润高速增长 | +| 市场范围 | 沪深A股 | 仅限A股市场 | +| ST股票 | 排除 | 避免风险股票 | +| 创业板 | 排除 | 聚焦主板和中小板 | +| 科创板 | 排除 | 避免高估值板块 | + +### 排序规则 + +按**总市值由小到大**排序，优先选择最小盘的股票。 + +## 💰 量化交易策略 + +### 资金管理 + +| 参数 | 设置 | 说明 | +|------|------|------| +| 资金量 | 10万元 | 初始投资金额 | +| 交易仓位 | 满仓 | 100%仓位运作 | +| 个股最大持仓 | 30% | 单只股票最多3万元 | +| 账户最大持股数 | 4只 | 分散风险 | +| 单日最大买入数 | 2只 | 控制买入节奏 | + +### 买卖时机 + +**买入时机**： +- 开盘买入 +- 按总市值由小到大排序 +- 优先买入市值更小的股票 + +**卖出时机**： +1. **持股到期**：持股满5天自动提醒卖出 +2. **技术信号**：MA5下穿MA20时提醒卖出 + +## 📊 策略监控 + +### 监控功能 + +小市值策略复用"低价擒牛"的监控框架，提供完整的监控服务： + +- ✅ 自动跟踪持仓股票 +- ✅ 每分钟扫描一次（可配置） +- ✅ 计算MA5和MA20均线 +- ✅ 双重卖出信号检测 +- ✅ 钉钉自动通知 +- ✅ 自动移除已触发股票 + +### 使用流程 + +``` +1. 执行选股 + ↓ +2. 查看筛选结果 + ↓ +3. 点击"➕ 加入策略监控" + ↓ +4. 进入"📊 策略监控"面板 + ↓ +5. 统一管理所有监控股票 + ↓ +6. 收到卖出提醒 + ↓ +7. 执行卖出操作 +``` + +## 🔔 通知功能 + +### 选股通知 + +选股完成后可以发送钉钉通知，包含： +- 筛选策略说明 +- 筛选数量 +- 精选股票列表 +- 生成时间 + +### 卖出提醒 + +触发卖出条件时自动发送通知： +- 股票代码和名称 +- 提醒类型（持股到期/MA死叉） +- 提醒原因 +- 当前价格和MA数据 +- 持有天数 +- 卖出建议 + +## 💡 策略优势 + +### 1. 小盘优势 +- **成长空间大**：小市值股票有更大的成长空间 +- **灵活性强**：容易被市场关注和炒作 +- **爆发力强**：小盘股涨幅潜力更大 + +### 2. 高成长性 +- **营收增长**：营收增长率≥10%保证业务增长 +- **利润爆发**：净利润增长率≥100%筛选高速成长股 +- **双重保障**：营收和利润双重验证 + +### 3. 风险控制 +- **排除ST**：避免财务风险股票 +- **排除高估值板块**：不选创业板和科创板 +- **仓位管理**：单股最多30%，最多持有4只 + +### 4. 智能监控 +- **自动跟踪**：无需人工盯盘 +- **双重信号**：时间+技术双重保护 +- **即时通知**：钉钉消息实时接收 + +## 📋 与低价擒牛的对比 + +| 特性 | 低价擒牛 | 小市值策略 | +|------|---------|-----------| +| **筛选重点** | 股价 | 市值 | +| **价格条件** | <10元 | 无限制 | +| **市值条件** | 无限制 | ≤50亿 | +| **营收要求** | 无 | ≥10% | +| **净利增长** | ≥100% | ≥100% | +| **板块限制** | 深圳A股 | 沪深A股，非创业/科创 | +| **资金量** | 100万 | 10万 | +| **个股仓位** | 40% | 30% | +| **最大持股** | 4只 | 4只 | + +## 🚀 使用示例 + +### 步骤1：选股 + +1. 点击左侧菜单"📊 小市值策略" +2. 设置筛选数量（默认5只） +3. 点击"🚀 开始小市值策略选股" +4. 等待筛选结果 + +### 步骤2：查看结果 + +筛选结果显示： +- 股票代码和名称 +- 总市值 +- 营收增长率 +- 净利润增长率 +- 所属行业 +- 其他财务指标 + +### 步骤3：加入监控 + +1. 在股票详情中点击"➕ 加入策略监控" +2. 系统自动记录买入信息 +3. 提示"已加入监控列表" + +### 步骤4：监控管理 + +1. 点击右上角"📊 策略监控" +2. 查看所有监控中的股票 +3. 启动监控服务 +4. 等待卖出提醒 + +### 步骤5：接收通知 + +- 钉钉收到卖出提醒 +- 查看提醒详情 +- 执行卖出操作 + +## ⚙️ 配置说明 + +### 环境变量 + +小市值策略使用与低价擒牛相同的配置： + +```bash +# .env文件 +TDX_BASE_URL=http://192.168.1.222:8181 # TDX API地址 +LOW_PRICE_BULL_SCAN_INTERVAL=60 # 扫描间隔（秒） +LOW_PRICE_BULL_HOLDING_DAYS=5 # 持股天数限制 + +# Webhook通知 +WEBHOOK_ENABLED=true +WEBHOOK_URL=your_dingtalk_webhook_url +WEBHOOK_TYPE=dingtalk +WEBHOOK_KEYWORD=aiagents通知 +``` + +### 监控服务 + +小市值策略和低价擒牛共享同一个监控服务： +- 统一的监控面板 +- 统一的数据库 +- 统一的提醒机制 + +## ⚠️ 注意事项 + +### 1. 风险提示 +- 小盘股流动性相对较差 +- 波动性较大 +- 需要设置止损 + +### 2. 数据要求 +- 需要TDX API支持 +- 至少20天K线数据 +- 实时数据更新 + +### 3. 监控服务 +- 需要手动启动 +- 保持系统运行 +- 定期检查状态 + +### 4. 策略执行 +- 建议分批买入 +- 严格执行止损 +- 不要追涨杀跌 + +## 📈 策略逻辑 + +### 选股逻辑 + +``` +总市值≤50亿 + ↓ +营收增长率≥10% + ↓ +净利润增长率≥100% + ↓ +沪深A股 + ↓ +排除ST/创业板/科创板 + ↓ +按总市值排序 + ↓ +选择前N只 +``` + +### 监控逻辑 + +``` +加入监控 + ↓ +记录买入信息 + ↓ +每分钟扫描 + ↓ +检查持股天数 + ↓ +检查MA5/MA20 + ↓ +触发卖出条件？ + ↓ +生成提醒 + ↓ +发送通知 + ↓ +自动移除 +``` + +## 🎯 适用场景 + +### 适合 +- ✅ 寻找高成长小盘股 +- ✅ 短线波段操作 +- ✅ 资金量较小的投资者 +- ✅ 愿意承担一定风险 + +### 不适合 +- ❌ 追求稳健收益 +- ❌ 长期价值投资 +- ❌ 风险承受能力低 +- ❌ 对流动性要求高 + +## 📚 相关文档 + +- [低价擒牛策略监控配置说明](./低价擒牛策略监控配置说明.md) +- [TDX API配置快速指南](./TDX_API配置快速指南.md) +- [Webhook通知配置指南](./Webhook通知配置指南.md) + +--- + +**更新时间**：2024-12-14 +**版本**：v1.0 diff --git a/small_cap_selector.py b/small_cap_selector.py new file mode 100644 index 0000000..bb03c2d --- /dev/null +++ b/small_cap_selector.py @@ -0,0 +1,140 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +小市值策略选股模块 +使用pywencai进行股票筛选 +""" + +import logging +from typing import Tuple, Optional +import pandas as pd + + +class SmallCapSelector: + """小市值策略选股器""" + + def __init__(self): + self.logger = logging.getLogger(__name__) + + def get_small_cap_stocks(self, top_n: int = 5) -> Tuple[bool, Optional[pd.DataFrame], str]: + """ + 获取符合小市值策略的股票 + + 筛选条件： + - 总市值 ≤ 50亿 + - 营收增长率 ≥ 10% + - 净利润增长率 ≥ 100% + - 沪深A股 + - 非ST + - 非创业板 + - 非科创板 + - 按总市值由小到大排名 + + Args: + top_n: 返回前N只股票 + + Returns: + (是否成功, 数据DataFrame, 消息) + """ + try: + import pywencai + + # 构建查询语句（按总市值由小至大排名） + query = ( + "总市值≤50亿，" + "营收增长率≥10%，" + "净利润增长率(净利润同比增长率)≥100%，" + "沪深A股，" + "非ST，" + "非创业板，" + "非科创板，" + "总市值由小至大排名" + ) + + self.logger.info(f"开始执行小市值策略选股，查询条件: {query}") + + # 调用pywencai + result = pywencai.get(query=query, loop=True) + + if result is None or result.empty: + self.logger.warning("未获取到符合条件的股票") + return False, None, "未找到符合条件的股票" + + self.logger.info(f"获取到 {len(result)} 只股票") + + # 取前N只 + if len(result) > top_n: + result = result.head(top_n) + self.logger.info(f"筛选前 {top_n} 只股票") + + return True, result, f"成功获取 {len(result)} 只股票" + + except ImportError: + error_msg = "pywencai模块未安装，请执行: pip install pywencai" + self.logger.error(error_msg) + return False, None, error_msg + + except Exception as e: + error_msg = f"选股失败: {str(e)}" + self.logger.error(error_msg) + import traceback + traceback.print_exc() + return False, None, error_msg + + def format_stock_info(self, df: pd.DataFrame) -> str: + """ + 格式化股票信息为文本 + + Args: + df: 股票数据DataFrame + + Returns: + 格式化后的文本 + """ + if df is None or df.empty: + return "无数据" + + lines = [] + for idx, row in df.iterrows(): + stock_code = row.get('股票代码', 'N/A') + stock_name = row.get('股票简称', 'N/A') + market_cap = row.get('总市值', row.get('总市值[20241211]', 'N/A')) + revenue_growth = row.get('营收增长率', row.get('营业收入增长率', 'N/A')) + profit_growth = row.get('净利润增长率', row.get('净利润同比增长率', 'N/A')) + + line = f"{idx+1}. {stock_code} {stock_name}" + + # 添加详细信息 + details = [] + if market_cap != 'N/A': + try: + cap_val = float(market_cap) + if cap_val >= 100000000: + details.append(f"市值:{cap_val/100000000:.2f}亿") + else: + details.append(f"市值:{cap_val/10000:.2f}万") + except: + pass + + if revenue_growth != 'N/A': + try: + details.append(f"营收增长:{float(revenue_growth):.2f}%") + except: + pass + + if profit_growth != 'N/A': + try: + details.append(f"净利增长:{float(profit_growth):.2f}%") + except: + pass + + if details: + line += f" - {', '.join(details)}" + + lines.append(line) + + return '\n'.join(lines) + + +# 全局实例 +small_cap_selector = SmallCapSelector() diff --git a/small_cap_ui.py b/small_cap_ui.py new file mode 100644 index 0000000..690d403 --- /dev/null +++ b/small_cap_ui.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +小市值策略UI模块 +""" + +import streamlit as st +import pandas as pd +from datetime import datetime +from small_cap_selector import small_cap_selector +from notification_service import notification_service +from low_price_bull_monitor import low_price_bull_monitor +from low_price_bull_service import low_price_bull_service + + +def display_small_cap(): + """显示小市值策略界面""" + + # 检查是否显示监控面板 + if st.session_state.get('show_small_cap_monitor'): + from low_price_bull_monitor_ui import display_monitor_panel + display_monitor_panel() + + # 返回按钮 + if st.button("🔙 返回选股", type="secondary"): + del st.session_state.show_small_cap_monitor + st.rerun() + return + + st.markdown("顶部按钮区") + col_select, col_monitor = st.columns([3, 1]) + + with col_select: + st.markdown("## 📊 小市值策略 - 小盘高成长股票筛选") + + with col_monitor: + st.write("") # 占位 + if st.button("📊 策略监控", type="primary", use_container_width=True): + st.session_state.show_small_cap_monitor = True + st.rerun() + + st.markdown("---") + + st.markdown(""" + ### 📋 选股策略说明 + + **筛选条件**： + - ✅ 总市值 ≤ 50亿 + - ✅ 营收增长率 ≥ 10% + - ✅ 净利润增长率 ≥ 100%（净利润同比增长率） + - ✅ 沪深A股 + - ✅ 非ST股票 + - ✅ 非创业板 + - ✅ 非科创板 + - ✅ 按总市值由小至大排名 + + **量化交易策略**： + - 💰 资金量：10万元 + - 📅 持股周期：5天 + - 💼 仓位控制：满仓 + - 📊 个股最大持仓：3成（30%） + - 🎯 账户最大持股数：4只 + - 🛒 单日最大买入数：2只 + - 📈 买入时机：开盘买入 + - 📉 卖出时机：MA5下穿MA20或持股满5天 + """) + + st.markdown("---") + + # 参数设置 + col1, col2 = st.columns([2, 1]) + + with col1: + top_n = st.slider( + "筛选数量", + min_value=3, + max_value=10, + value=5, + step=1, + help="选择展示的股票数量" + ) + + with col2: + st.info(f"💡 将筛选市值最小的前{top_n}只股票") + + st.markdown("---") + + # 开始选股按钮 + if st.button("🚀 开始小市值策略选股", type="primary", use_container_width=True): + + with st.spinner("正在获取数据，请稍候..."): + # 创建选股器 + success, stocks_df, message = small_cap_selector.get_small_cap_stocks(top_n) + + if not success: + st.error(f"❌ {message}") + return + + st.success(f"✅ {message}") + + # 保存到session_state + st.session_state.small_cap_stocks = stocks_df + st.session_state.small_cap_time = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + # 显示选股结果 + if 'small_cap_stocks' in st.session_state and st.session_state.small_cap_stocks is not None: + st.markdown("---") + st.markdown("## 📈 选股结果") + + stocks_df = st.session_state.small_cap_stocks + select_time = st.session_state.small_cap_time + + st.info(f"🕒 选股时间：{select_time} | 📊 股票数量：{len(stocks_df)} 只") + + # 显示股票列表 + display_stock_list(stocks_df) + + # 发送钉钉通知 + st.markdown("---") + if st.button("📲 发送钉钉通知", type="secondary", use_container_width=True): + send_dingtalk_notification(stocks_df) + + +def display_stock_list(stocks_df: pd.DataFrame): + """显示股票列表""" + + for idx, row in stocks_df.iterrows(): + stock_code = row.get('股票代码', 'N/A') + stock_name = row.get('股票简称', 'N/A') + + with st.expander(f"📊 {idx+1}. {stock_code} {stock_name}", expanded=True): + display_stock_detail(row) + + +def display_stock_detail(row: pd.Series): + """显示股票详细信息""" + + # 获取所有可能的字段 + financial_fields = [ + ('总市值', row.get('总市值', row.get('总市值[20241211]', None))), + ('营收增长率', row.get('营收增长率', row.get('营业收入增长率', None))), + ('净利润增长率', row.get('净利润增长率', row.get('净利润同比增长率', None))), + ('股价', row.get('股价', row.get('最新价', None))), + ('市盈率', row.get('市盈率', row.get('市盈率TTM', None))), + ('市净率', row.get('市净率', row.get('市净率PB', None))), + ('所属行业', row.get('所属行业', row.get('所属同花顺行业', None))), + ] + + # 检查是否有任何有效数据 + has_any_data = any(is_valid_value(value) for _, value in financial_fields) + + # 决定布局 + if has_any_data: + col1, col2 = st.columns(2) + else: + col1 = st.container() + col2 = None + + with col1: + st.markdown("#### 📊 基本信息") + st.markdown(f"**股票代码**: {row.get('股票代码', 'N/A')}") + st.markdown(f"**股票名称**: {row.get('股票简称', 'N/A')}") + + # 只有当有财务数据时才显示财务指标 + if col2 is not None: + with col2: + st.markdown("#### 💼 财务指标") + + for field_name, value in financial_fields: + if is_valid_value(value): + formatted_value = format_value(value, get_suffix(field_name)) + st.markdown(f"**{field_name}**: {formatted_value}") + + # 添加监控按钮 + st.markdown("---") + st.markdown("#### 📊 策略监控") + + from low_price_bull_monitor_ui import add_stock_to_monitor_button + + stock_code = row.get('股票代码', '') + stock_name = row.get('股票简称', '') + price = row.get('股价', row.get('最新价', None)) + + # 去掉代码后缀 + if isinstance(stock_code, str) and '.' in stock_code: + stock_code = stock_code.split('.')[0] + + # 转换价格 + try: + price_float = float(price) if price and not pd.isna(price) else None + except: + price_float = None + + if stock_code and stock_name: + add_stock_to_monitor_button(stock_code, stock_name, price_float) + + +def is_valid_value(value): + """判断值是否有效""" + if value is None: + return False + if pd.isna(value): + return False + if str(value).strip() in ['', 'N/A', 'nan', 'None']: + return False + return True + + +def format_value(value, suffix=''): + """格式化显示值""" + if isinstance(value, (int, float)): + if abs(value) >= 100000000: # 亿 + return f"{value/100000000:.2f}亿{suffix}" + elif abs(value) >= 10000: # 万 + return f"{value/10000:.2f}万{suffix}" + else: + return f"{value:.2f}{suffix}" + return f"{value}{suffix}" + + +def get_suffix(field_name: str) -> str: + """获取字段后缀""" + suffix_map = { + '总市值': '元', + '股价': '元', + '营收增长率': '%', + '净利润增长率': '%', + } + return suffix_map.get(field_name, '') + + +def send_dingtalk_notification(stocks_df: pd.DataFrame): + """发送钉钉通知""" + + try: + if not notification_service.config['webhook_enabled']: + st.warning("⚠️ Webhook通知未启用，请在系统配置中启用") + return + + # 构建消息 + keyword = notification_service.config.get('webhook_keyword', 'aiagents通知') + + message_text = f"### {keyword} - 小市值策略选股完成\n\n" + message_text += "**筛选策略**: 总市值≤50亿 + 营收增长率≥10% + 净利润增长率≥100% + 沪深A股\n\n" + message_text += f"**筛选数量**: {len(stocks_df)} 只\n\n" + message_text += "**精选股票**:\n\n" + + for idx, row in stocks_df.iterrows(): + stock_code = row.get('股票代码', 'N/A') + stock_name = row.get('股票简称', 'N/A') + message_text += f"{idx+1}. {stock_code} {stock_name}\n\n" + + message_text += f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n" + message_text += "_此消息由AI股票分析系统自动发送_" + + # 直接发送钉钉Webhook + if notification_service.config['webhook_type'] == 'dingtalk': + import requests + + data = { + "msgtype": "markdown", + "markdown": { + "title": f"{keyword}", + "text": message_text + } + } + + response = requests.post( + notification_service.config['webhook_url'], + json=data, + headers={'Content-Type': 'application/json'}, + timeout=10 + ) + + if response.status_code == 200: + st.success("✅ 钉钉通知发送成功") + else: + st.error(f"❌ 钉钉通知发送失败: HTTP {response.status_code}") + else: + st.warning("⚠️ 当前仅支持钉钉通知") + + except Exception as e: + st.error(f"❌ 发送通知失败: {str(e)}")