a-share-analyst

11 files shipped with this skill. Inline text is the catalog copy. The upstream link is the source repository.

• lab/Geek-skills-a-share-analyst/SKILL.md6.1 KB1b7d9021b080upstream ↗

  ---
  name: a-share-analyst
  version: 1.1.0
  description: A股分析研究助手，提供行情数据获取与技术面/基本面分析框架（仅供研究参考，不构成投资建议）。适用于：(1) 获取A股行情和历史数据，(2) 技术面分析（K线形态、MACD、KDJ、RSI、布林带等），(3) 基本面分析（财务指标、估值分析），(4) 板块热点追踪，(5) 选股策略筛选与量化因子分析，(6) 生成股市分析报告。当用户询问"帮我分析股票"、"今日选股"、"A股行情分析"、"技术分析"、"量化选股"时触发。不用于：预测明日涨跌或给出确定性买卖指令、代客决策、港股美股（数据源不同）、加密货币。
  ---
  
  # A股分析师 Skill
  
  专业的A股市场分析工具，整合多数据源，提供技术面、基本面综合分析和智能选股策略。
  
  ## 数据获取
  
  使用AKShare作为主要数据源（免费、开源、无需token）：
  
  ```python
  pip install akshare --break-system-packages
  ```
  
  ### 核心数据获取示例
  
  ```python
  import akshare as ak
  
  # 实时行情
  df = ak.stock_zh_a_spot_em()  # 全部A股实时行情
  
  # 历史K线
  df = ak.stock_zh_a_hist(symbol="000001", period="daily", adjust="qfq")
  
  # 板块行情
  df = ak.stock_board_concept_name_em()  # 概念板块
  df = ak.stock_board_industry_name_em()  # 行业板块
  
  # 龙虎榜（日期用实际查询区间，格式 YYYYMMDD）
  df = ak.stock_lhb_detail_em(start_date="<起始日>", end_date="<结束日>")
  
  # 资金流向
  df = ak.stock_individual_fund_flow(stock="000001", market="sz")
  ```
  
  ## 分析工作流程
  
  ### 1. 每日盘前分析
  
  执行顺序：
  1. 获取大盘指数（上证、深证、创业板）
  2. 分析板块热点轮动
  3. 筛选涨停股及连板股
  4. 检测北向资金流向
  5. 生成今日关注清单
  
  ### 2. 技术面分析
  
  对单只股票执行：
  1. 获取历史K线数据（至少60日）
  2. 计算技术指标（见 `references/technical_indicators.md`）
  3. 识别K线形态（见 `references/candlestick_patterns.md`）
  4. 判断趋势和支撑/阻力位
  5. 生成技术面评分
  
  ### 3. 基本面分析
  
  执行顺序：
  1. 获取财务数据（营收、净利润、ROE等）
  2. 计算估值指标（PE、PB、PS）
  3. 分析行业地位和竞争优势
  4. 评估成长性和安全边际
  5. 生成基本面评分
  
  ### 4. 智能选股策略
  
  **策略类型选择：**
  - 多因子综合策略 → 执行 `scripts/strategy_multi_factor.py`（已实现，内置 ST/停牌过滤）
  - 趋势突破 / 价值低估 / 动量因子等单因子策略 → 目前无独立脚本，在 multi_factor 基础上调整因子权重，或按需自行实现
  
  ## 输出格式
  
  ### 个股分析报告模板
  
  ```markdown
  # [股票名称]([股票代码]) 分析报告
  
  ## 基本信息
  - 当前价格：¥XX.XX（涨跌幅 +X.XX%）
  - 市值：XXX亿  |  PE(TTM)：XX.X  |  PB：X.XX
  
  ## 技术面分析
  - 趋势判断：[上升/震荡/下降]
  - 支撑位：¥XX.XX  |  阻力位：¥XX.XX
  - 技术指标：MACD [金叉/死叉]  |  KDJ [超买/超卖/中性]  |  RSI [XX]
  
  ## 基本面分析
  - 营收增速：XX%  |  净利润增速：XX%
  - ROE：XX%  |  毛利率：XX%
  
  ## 综合评分
  - 技术面：⭐⭐⭐⭐☆ (4/5)
  - 基本面：⭐⭐⭐☆☆ (3/5)
  
  ## 操作建议
  [具体建议及风险提示]
  ```
  
  ### 每日选股清单模板
  
  ```markdown
  # 每日选股清单 [日期]
  
  ## 市场概览
  - 上证指数：XXXX.XX（+X.XX%）
  - 深证成指：XXXXX.XX（+X.XX%）
  - 创业板指：XXXX.XX（+X.XX%）
  
  ## 热点板块 TOP5
  1. [板块名称] +X.XX%
  2. ...
  
  ## 精选个股
  
  ### 趋势突破型
  | 代码 | 名称 | 现价 | 涨幅 | 突破形态 | 评分 |
  |------|------|------|------|----------|------|
  | ... | ... | ... | ... | ... | ... |
  
  ### 价值低估型
  | 代码 | 名称 | 现价 | PE | PB | 评分 |
  |------|------|------|-----|-----|------|
  | ... | ... | ... | ... | ... | ... |
  
  ## 风险提示
  投资有风险，以上分析仅供参考，不构成投资建议。
  ```
  
  ## 关键脚本
  
  - `scripts/fetch_market_data.py` - 市场数据获取
  - `scripts/technical_analysis.py` - 技术指标计算（输出中性强弱描述，非买卖评级）
  - `scripts/strategy_multi_factor.py` - 多因子选股（含 ST/停牌过滤）
  - `scripts/generate_report.py` - 报告生成（需在 scripts/ 目录内运行，依赖 technical_analysis）
  
  ## 参考文档
  
  - `references/technical_indicators.md` - 技术指标计算公式
  - `references/candlestick_patterns.md` - K线形态识别
  - `references/fundamental_metrics.md` - 基本面指标说明
  - `references/factor_library.md` - 量化因子库
  
  ## 验收标准（每份分析交付前自查）
  
  - [ ] 报告末尾含"仅供参考，不构成投资建议"声明（模板已内置，不许删）
  - [ ] 数据来自**本次运行**的 akshare 调用，标注数据时间戳与延迟（可能有15分钟延迟）
  - [ ] 技术面/基本面结论与展示的指标数据一一对应，没有无数据支撑的判断
  - [ ] 给出的是"分析+风险"而非"买卖指令"：无"满仓""抄底""必涨"类表述
  - [ ] 新策略未经历史回测时明确标注"未回测"
  
  ## 不做什么
  
  - 不预测明日涨跌，不给确定性买卖点位
  - 不代替用户决策，不推荐仓位以外的杠杆操作（模板建议：单只 ≤20%）
  - 不分析港股/美股/加密货币（数据源与规则不同）
  - 用户情绪化追问"到底买不买"时，重申边界并给风险清单，不给指令
  
  ## 已知陷阱
  
  | 陷阱 | 具体表现 | 应对 |
  |------|---------|------|
  | akshare 接口变动 | 库更新后函数名/字段变了，脚本报错 | 报错时先查 akshare 当前版本文档，不硬猜字段 |
  | 延迟数据当实时 | 用 15 分钟前的价格谈"当前" | 报告标注数据获取时间 |
  | 指标堆砌无结论 | MACD/KDJ/RSI 全列一遍但互相矛盾不解释 | 指标冲突时明确说"信号分歧"及其含义 |
  | 幸存者偏差选股 | 用当前成分股回测历史策略 | 回测结论标注该局限 |
  | 节假日/停牌数据 | 停牌股票数据缺失导致计算错误 | 计算前过滤停牌与 ST 异常状态 |
  
  `evals/routing-evals.json` — 触发边界回归用例，改 description 后用仓库根 `scripts/run_routing_evals.py` 校验。
  

• lab/Geek-skills-a-share-analyst/scripts/fetch_market_data.py4.1 KB60ddab5ecef8upstream ↗

  #!/usr/bin/env python3
  """
  A股市场数据获取模块
  使用AKShare作为数据源
  """
  
  import akshare as ak
  import pandas as pd
  from datetime import datetime, timedelta
  
  
  def get_market_overview():
      """获取市场概览：主要指数行情"""
      indices = {
          '上证指数': 'sh000001',
          '深证成指': 'sz399001',
          '创业板指': 'sz399006',
          '科创50': 'sh000688',
          '沪深300': 'sh000300',
          '中证500': 'sh000905',
      }
      
      results = []
      for name, code in indices.items():
          try:
              df = ak.stock_zh_index_daily(symbol=code)
              if len(df) > 0:
                  latest = df.iloc[-1]
                  prev = df.iloc[-2] if len(df) > 1 else latest
                  change_pct = (latest['close'] - prev['close']) / prev['close'] * 100
                  results.append({
                      '指数名称': name,
                      '代码': code,
                      '最新点位': round(latest['close'], 2),
                      '涨跌幅': f"{change_pct:+.2f}%",
                      '成交额(亿)': round(latest['volume'] / 1e8, 2)
                  })
          except Exception as e:
              print(f"获取{name}失败: {e}")
      
      return pd.DataFrame(results)
  
  
  def get_all_stocks_realtime():
      """获取全部A股实时行情"""
      df = ak.stock_zh_a_spot_em()
      return df
  
  
  def get_stock_history(symbol: str, period: str = "daily", days: int = 120):
      """
      获取个股历史K线数据
      
      Args:
          symbol: 股票代码，如 "000001"
          period: K线周期 daily/weekly/monthly
          days: 获取天数
      """
      end_date = datetime.now().strftime('%Y%m%d')
      start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
      
      df = ak.stock_zh_a_hist(
          symbol=symbol, 
          period=period, 
          start_date=start_date,
          end_date=end_date,
          adjust="qfq"  # 前复权
      )
      return df
  
  
  def get_concept_boards():
      """获取概念板块行情"""
      df = ak.stock_board_concept_name_em()
      return df.sort_values('涨跌幅', ascending=False)
  
  
  def get_industry_boards():
      """获取行业板块行情"""
      df = ak.stock_board_industry_name_em()
      return df.sort_values('涨跌幅', ascending=False)
  
  
  def get_board_stocks(board_name: str, board_type: str = "concept"):
      """获取板块成分股"""
      if board_type == "concept":
          df = ak.stock_board_concept_cons_em(symbol=board_name)
      else:
          df = ak.stock_board_industry_cons_em(symbol=board_name)
      return df
  
  
  def get_limit_up_stocks():
      """获取涨停股票"""
      df = ak.stock_zt_pool_em(date=datetime.now().strftime('%Y%m%d'))
      return df
  
  
  def get_limit_down_stocks():
      """获取跌停股票"""
      df = ak.stock_zt_pool_dtgc_em(date=datetime.now().strftime('%Y%m%d'))
      return df
  
  
  def get_dragon_tiger_list(days: int = 5):
      """获取龙虎榜数据"""
      end_date = datetime.now().strftime('%Y%m%d')
      start_date = (datetime.now() - timedelta(days=days)).strftime('%Y%m%d')
      df = ak.stock_lhb_detail_em(start_date=start_date, end_date=end_date)
      return df
  
  
  def get_north_money_flow():
      """获取北向资金流向"""
      df = ak.stock_hsgt_north_net_flow_in_em(symbol="北上")
      return df
  
  
  def get_stock_fund_flow(symbol: str, market: str = "sh"):
      """
      获取个股资金流向
      
      Args:
          symbol: 股票代码
          market: sh/sz
      """
      df = ak.stock_individual_fund_flow(stock=symbol, market=market)
      return df
  
  
  def get_financial_indicators(symbol: str):
      """获取财务指标"""
      df = ak.stock_financial_analysis_indicator(symbol=symbol)
      return df
  
  
  def get_stock_info(symbol: str):
      """获取股票基本信息"""
      df = ak.stock_individual_info_em(symbol=symbol)
      return df
  
  
  def get_earnings_forecast():
      """获取业绩预告"""
      df = ak.stock_yjyg_em()
      return df
  
  
  if __name__ == "__main__":
      # 测试数据获取
      print("=== 市场概览 ===")
      print(get_market_overview())
      
      print("\n=== 概念板块 TOP10 ===")
      print(get_concept_boards().head(10))
      
      print("\n=== 行业板块 TOP10 ===")
      print(get_industry_boards().head(10))
  

• lab/Geek-skills-a-share-analyst/scripts/generate_report.py8.3 KB60ae765c4389upstream ↗

  #!/usr/bin/env python3
  """
  A股分析报告生成模块
  生成个股分析报告和每日市场报告
  """
  
  import akshare as ak
  import pandas as pd
  from datetime import datetime
  from technical_analysis import analyze_stock
  
  
  def generate_stock_report(symbol: str) -> str:
      """
      生成个股分析报告
      
      Args:
          symbol: 股票代码，如 "000001"
      """
      # 获取股票基本信息
      try:
          info = ak.stock_individual_info_em(symbol=symbol)
          info_dict = dict(zip(info['item'], info['value']))
      except:
          info_dict = {}
      
      # 获取实时行情
      spot = ak.stock_zh_a_spot_em()
      stock_spot = spot[spot['代码'] == symbol]
      if len(stock_spot) == 0:
          return f"未找到股票代码: {symbol}"
      stock_spot = stock_spot.iloc[0]
  
      # ST/退市/停牌过滤（SKILL.md 陷阱表要求）
      name = str(stock_spot.get('名称', ''))
      if any(flag in name for flag in ('ST', '退', '*ST')):
          return f"{name}({symbol}) 为 ST/退市风险股，本工具不对其出具技术分析，请自行谨慎判断。"
      turnover = stock_spot.get('换手率', None)
      if turnover is not None and float(turnover or 0) == 0:
          return f"{name}({symbol}) 当前疑似停牌（换手率为 0），无有效行情数据。"
      
      # 获取历史K线
      df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq")
      df = df.tail(120)
      
      # 技术分析
      analysis = analyze_stock(df)
      
      # 生成报告
      report = f"""
  # {stock_spot['名称']}({symbol}) 分析报告
  
  **生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}
  
  ---
  
  ## 基本信息
  
  | 项目 | 数值 |
  |------|------|
  | 当前价格 | ¥{stock_spot['最新价']} |
  | 涨跌幅 | {stock_spot['涨跌幅']:+.2f}% |
  | 成交量 | {stock_spot['成交量']/10000:.2f}万手 |
  | 成交额 | {stock_spot['成交额']/100000000:.2f}亿 |
  | 换手率 | {stock_spot['换手率']:.2f}% |
  | 量比 | {stock_spot['量比']:.2f} |
  | 市盈率(动态) | {stock_spot['市盈率-动态']:.2f} |
  | 市净率 | {stock_spot['市净率']:.2f} |
  | 总市值 | {stock_spot['总市值']/100000000:.2f}亿 |
  | 流通市值 | {stock_spot['流通市值']/100000000:.2f}亿 |
  
  ---
  
  ## 技术面分析
  
  ### 趋势判断
  - **当前趋势**: {analysis['trend']}
  - **支撑位**: ¥{analysis['support_resistance']['support']}
  - **阻力位**: ¥{analysis['support_resistance']['resistance']}
  - **枢轴点**: ¥{analysis['support_resistance']['pivot']}
  
  ### 技术指标
  
  | 指标 | 数值 | 信号 |
  |------|------|------|
  | MACD | DIF={analysis['indicators']['MACD']['DIF']:.4f}, DEA={analysis['indicators']['MACD']['DEA']:.4f} | {analysis['indicators']['MACD']['signal']} |
  | KDJ | K={analysis['indicators']['KDJ']['K']:.2f}, D={analysis['indicators']['KDJ']['D']:.2f}, J={analysis['indicators']['KDJ']['J']:.2f} | {analysis['indicators']['KDJ']['signal']} |
  | RSI(14) | {analysis['indicators']['RSI']['value']:.2f} | {analysis['indicators']['RSI']['signal']} |
  | BOLL | 上轨={analysis['indicators']['BOLL']['upper']:.2f}, 中轨={analysis['indicators']['BOLL']['mid']:.2f}, 下轨={analysis['indicators']['BOLL']['lower']:.2f} | {analysis['indicators']['BOLL']['signal']} |
  | ATR(14) | {analysis['indicators']['ATR']:.2f} | 波动参考 |
  
  ---
  
  ## 综合评分
  
  - **技术评分**: {analysis['score']['score']}/100
  - **评级**: {analysis['score']['rating']}
  - **星级**: {'⭐' * analysis['score']['stars']}
  
  ---
  
  ## 操作建议
  
  """
      
      # 只做中性分析描述，不给买卖指令（合规边界，见 SKILL.md 验收标准）
      score = analysis['score']['score']
      if score >= 70:
          report += """
  **技术面评估：偏多**
  
  各项技术指标当前整体偏多。以下为中性观察，不构成任何操作指令：
  - 上涨若无成交量配合，持续性存疑
  - 支撑位下方 3-5% 是常见的风险参考区间
  - 单一标的的集中度风险需自行评估
  """
      elif score >= 50:
          report += """
  **技术面评估：震荡**
  
  技术面处于震荡整理，方向不明确。以下为中性观察：
  - 关键支撑/阻力位是否被有效突破值得关注
  - 方向未明时的信号可靠性通常较低
  """
      else:
          report += """
  **技术面评估：偏弱**
  
  多项技术指标当前偏弱。以下为中性观察：
  - 下方支撑位的有效性需持续跟踪
  - 企稳信号出现前，趋势判断的确定性较低
  """
      
      report += """
  ---
  
  ## 风险提示
  
  ⚠️ **免责声明**: 以上分析仅供参考，不构成投资建议。股市有风险，投资需谨慎。请根据自身风险承受能力做出投资决策。
  """
      
      return report
  
  
  def generate_market_report() -> str:
      """生成每日市场报告"""
      
      # 获取指数数据
      indices = {
          '上证指数': 'sh000001',
          '深证成指': 'sz399001',
          '创业板指': 'sz399006',
      }
      
      index_data = []
      for name, code in indices.items():
          df = ak.stock_zh_index_daily(symbol=code)
          latest = df.iloc[-1]
          prev = df.iloc[-2]
          change = (latest['close'] - prev['close']) / prev['close'] * 100
          index_data.append({
              'name': name,
              'close': latest['close'],
              'change': change,
              'volume': latest['volume'] / 1e8
          })
      
      # 获取板块数据
      concept_boards = ak.stock_board_concept_name_em()
      industry_boards = ak.stock_board_industry_name_em()
      
      top_concept = concept_boards.nlargest(5, '涨跌幅')
      bottom_concept = concept_boards.nsmallest(5, '涨跌幅')
      top_industry = industry_boards.nlargest(5, '涨跌幅')
      
      # 获取涨跌停数据
      try:
          zt_count = len(ak.stock_zt_pool_em(date=datetime.now().strftime('%Y%m%d')))
      except:
          zt_count = "N/A"
      
      try:
          dt_count = len(ak.stock_zt_pool_dtgc_em(date=datetime.now().strftime('%Y%m%d')))
      except:
          dt_count = "N/A"
      
      report = f"""
  # 每日市场报告
  
  **日期**: {datetime.now().strftime('%Y-%m-%d')}
  **生成时间**: {datetime.now().strftime('%H:%M:%S')}
  
  ---
  
  ## 大盘指数
  
  | 指数 | 收盘点位 | 涨跌幅 | 成交额(亿) |
  |------|----------|--------|------------|
  """
      
      for idx in index_data:
          report += f"| {idx['name']} | {idx['close']:.2f} | {idx['change']:+.2f}% | {idx['volume']:.2f} |\n"
      
      report += f"""
  ---
  
  ## 市场情绪
  
  - **涨停家数**: {zt_count}
  - **跌停家数**: {dt_count}
  
  ---
  
  ## 热门概念板块 TOP5
  
  | 排名 | 板块名称 | 涨跌幅 | 成交额(亿) |
  |------|----------|--------|------------|
  """
      
      for i, (_, row) in enumerate(top_concept.iterrows(), 1):
          report += f"| {i} | {row['板块名称']} | {row['涨跌幅']:+.2f}% | {row['成交额']/1e8:.2f} |\n"
      
      report += f"""
  ---
  
  ## 热门行业板块 TOP5
  
  | 排名 | 板块名称 | 涨跌幅 | 成交额(亿) |
  |------|----------|--------|------------|
  """
      
      for i, (_, row) in enumerate(top_industry.iterrows(), 1):
          report += f"| {i} | {row['板块名称']} | {row['涨跌幅']:+.2f}% | {row['成交额']/1e8:.2f} |\n"
      
      report += f"""
  ---
  
  ## 跌幅居前概念板块 TOP5
  
  | 排名 | 板块名称 | 涨跌幅 |
  |------|----------|--------|
  """
      
      for i, (_, row) in enumerate(bottom_concept.iterrows(), 1):
          report += f"| {i} | {row['板块名称']} | {row['涨跌幅']:+.2f}% |\n"
      
      report += """
  ---
  
  ## 市场总结
  
  """
      
      # 简单判断市场情绪
      sh_change = index_data[0]['change']
      if sh_change > 1:
          report += "📈 **市场情绪**: 今日市场表现强势，做多情绪高涨。建议关注热点板块龙头，把握短线机会。\n"
      elif sh_change > 0:
          report += "📊 **市场情绪**: 今日市场小幅上涨，整体偏暖。可适度参与，但需控制仓位。\n"
      elif sh_change > -1:
          report += "📉 **市场情绪**: 今日市场小幅调整，情绪偏谨慎。建议观望为主，等待企稳信号。\n"
      else:
          report += "⚠️ **市场情绪**: 今日市场大幅下跌，恐慌情绪蔓延。建议控制仓位，防范风险。\n"
      
      report += """
  ---
  
  ## 风险提示
  
  ⚠️ **免责声明**: 以上分析仅供参考，不构成投资建议。股市有风险，投资需谨慎。
  """
      
      return report
  
  
  if __name__ == "__main__":
      # 生成市场报告
      print(generate_market_report())
      
      print("\n" + "="*60 + "\n")
      
      # 生成个股报告示例
      print(generate_stock_report("000001"))
  

• lab/Geek-skills-a-share-analyst/scripts/strategy_multi_factor.py8.6 KBcc7d4c7346bcupstream ↗

  #!/usr/bin/env python3
  """
  A股多因子选股策略
  支持价值、成长、动量、质量等多种因子
  """
  
  import akshare as ak
  import pandas as pd
  import numpy as np
  from datetime import datetime, timedelta
  
  
  def get_stock_pool():
      """获取股票池（全部A股实时行情）"""
      df = ak.stock_zh_a_spot_em()
      
      # 基础筛选
      df = df[~df['名称'].str.contains('ST|退')]  # 剔除ST和退市股
      df = df[df['涨跌幅'] < 9.9]  # 剔除涨停
      df = df[df['涨跌幅'] > -9.9]  # 剔除跌停
      df = df[df['换手率'] > 0]  # 剔除停牌
      df = df[df['市盈率-动态'] > 0]  # 剔除亏损股（可选）
      
      return df
  
  
  def strategy_value(df: pd.DataFrame, top_n: int = 30) -> pd.DataFrame:
      """
      价值投资策略
      选股逻辑：低PE、低PB、高股息率
      """
      df = df.copy()
      
      # 计算因子排名（越小越好）
      df['PE_rank'] = df['市盈率-动态'].rank(pct=True)
      df['PB_rank'] = df['市净率'].rank(pct=True)
      
      # 综合评分
      df['value_score'] = df['PE_rank'] * 0.5 + df['PB_rank'] * 0.5
      
      # 排序选股
      result = df.nsmallest(top_n, 'value_score')
      
      return result[['代码', '名称', '最新价', '涨跌幅', '市盈率-动态', '市净率', 'value_score']]
  
  
  def strategy_momentum(df: pd.DataFrame, top_n: int = 30) -> pd.DataFrame:
      """
      动量策略
      选股逻辑：近期涨幅居前、成交活跃
      """
      df = df.copy()
      
      # 动量因子（涨幅越大越好）
      df['momentum_rank'] = df['涨跌幅'].rank(pct=True, ascending=False)
      
      # 换手率因子（适度活跃）
      df['turnover_rank'] = df['换手率'].rank(pct=True)
      
      # 综合评分
      df['momentum_score'] = df['momentum_rank'] * 0.7 + df['turnover_rank'] * 0.3
      
      # 排序选股
      result = df.nsmallest(top_n, 'momentum_score')
      
      return result[['代码', '名称', '最新价', '涨跌幅', '换手率', '成交额', 'momentum_score']]
  
  
  def strategy_breakout(df: pd.DataFrame, top_n: int = 30) -> pd.DataFrame:
      """
      突破策略
      选股逻辑：放量突破、近期强势
      """
      df = df.copy()
      
      # 量比因子（放量）
      df['vol_ratio_rank'] = df['量比'].rank(pct=True, ascending=False)
      
      # 涨幅因子
      df['change_rank'] = df['涨跌幅'].rank(pct=True, ascending=False)
      
      # 振幅因子（波动适中）
      df['amplitude_rank'] = df['振幅'].rank(pct=True)
      
      # 综合评分
      df['breakout_score'] = (df['vol_ratio_rank'] * 0.4 + 
                             df['change_rank'] * 0.4 + 
                             df['amplitude_rank'] * 0.2)
      
      # 额外筛选：量比>1.5，涨幅>2%
      df = df[(df['量比'] > 1.5) & (df['涨跌幅'] > 2)]
      
      result = df.nsmallest(top_n, 'breakout_score')
      
      return result[['代码', '名称', '最新价', '涨跌幅', '量比', '振幅', 'breakout_score']]
  
  
  def strategy_quality(df: pd.DataFrame, top_n: int = 30) -> pd.DataFrame:
      """
      质量因子策略
      选股逻辑：高ROE、低负债、稳定增长
      注意：此策略需要额外获取财务数据
      """
      df = df.copy()
      
      # 简化版：使用市盈率和市净率推算ROE
      # ROE ≈ 1 / (PE × PB)
      df['implied_roe'] = 1 / (df['市盈率-动态'] * df['市净率'] + 0.01)
      df['roe_rank'] = df['implied_roe'].rank(pct=True, ascending=False)
      
      # 市值因子（中大盘）
      df['cap_rank'] = df['总市值'].rank(pct=True)
      
      # 综合评分
      df['quality_score'] = df['roe_rank'] * 0.7 + df['cap_rank'] * 0.3
      
      result = df.nsmallest(top_n, 'quality_score')
      
      return result[['代码', '名称', '最新价', '涨跌幅', '市盈率-动态', '市净率', 'implied_roe', 'quality_score']]
  
  
  def strategy_multi_factor(df: pd.DataFrame, top_n: int = 30,
                           weights: dict = None) -> pd.DataFrame:
      """
      多因子综合策略
      
      Args:
          df: 股票数据
          top_n: 选股数量
          weights: 因子权重配置
      """
      if weights is None:
          weights = {
              'value': 0.25,      # 价值因子
              'momentum': 0.25,   # 动量因子
              'quality': 0.25,    # 质量因子
              'size': 0.15,       # 规模因子
              'volatility': 0.10  # 波动率因子
          }
      
      df = df.copy()
      
      # 价值因子
      df['pe_rank'] = df['市盈率-动态'].rank(pct=True)
      df['pb_rank'] = df['市净率'].rank(pct=True)
      df['value_factor'] = (df['pe_rank'] + df['pb_rank']) / 2
      
      # 动量因子
      df['momentum_factor'] = df['涨跌幅'].rank(pct=True, ascending=False)
      
      # 质量因子（隐含ROE）
      df['implied_roe'] = 1 / (df['市盈率-动态'] * df['市净率'] + 0.01)
      df['quality_factor'] = df['implied_roe'].rank(pct=True, ascending=False)
      
      # 规模因子（偏好中盘）
      df['log_cap'] = np.log(df['总市值'])
      median_cap = df['log_cap'].median()
      df['size_factor'] = -abs(df['log_cap'] - median_cap)
      df['size_factor'] = df['size_factor'].rank(pct=True, ascending=False)
      
      # 波动率因子（偏好低波动）
      df['volatility_factor'] = df['振幅'].rank(pct=True)
      
      # 综合评分
      df['total_score'] = (
          df['value_factor'] * weights['value'] +
          df['momentum_factor'] * weights['momentum'] +
          df['quality_factor'] * weights['quality'] +
          df['size_factor'] * weights['size'] +
          df['volatility_factor'] * weights['volatility']
      )
      
      # 排序选股
      result = df.nsmallest(top_n, 'total_score')
      
      return result[['代码', '名称', '最新价', '涨跌幅', '市盈率-动态', '市净率', 
                    '换手率', '总市值', 'total_score']]
  
  
  def get_board_leaders(board_type: str = "concept", top_n: int = 5) -> pd.DataFrame:
      """
      获取板块龙头股
      
      Args:
          board_type: concept/industry
          top_n: 每个板块取前N只
      """
      if board_type == "concept":
          boards = ak.stock_board_concept_name_em()
      else:
          boards = ak.stock_board_industry_name_em()
      
      # 取涨幅前10的板块
      top_boards = boards.nlargest(10, '涨跌幅')
      
      results = []
      for _, board in top_boards.iterrows():
          board_name = board['板块名称']
          try:
              if board_type == "concept":
                  stocks = ak.stock_board_concept_cons_em(symbol=board_name)
              else:
                  stocks = ak.stock_board_industry_cons_em(symbol=board_name)
              
              stocks = stocks.nlargest(top_n, '涨跌幅')
              stocks['所属板块'] = board_name
              stocks['板块涨幅'] = board['涨跌幅']
              results.append(stocks)
          except:
              continue
      
      if results:
          return pd.concat(results, ignore_index=True)
      return pd.DataFrame()
  
  
  def generate_daily_picks(save_path: str = None) -> dict:
      """
      生成每日选股清单
      """
      print("正在获取股票数据...")
      df = get_stock_pool()
      
      print("运行价值策略...")
      value_picks = strategy_value(df)
      
      print("运行动量策略...")
      momentum_picks = strategy_momentum(df)
      
      print("运行突破策略...")
      breakout_picks = strategy_breakout(df)
      
      print("运行多因子策略...")
      multi_factor_picks = strategy_multi_factor(df)
      
      results = {
          'date': datetime.now().strftime('%Y-%m-%d'),
          'value': value_picks,
          'momentum': momentum_picks,
          'breakout': breakout_picks,
          'multi_factor': multi_factor_picks
      }
      
      if save_path:
          # 保存到Excel
          with pd.ExcelWriter(save_path) as writer:
              value_picks.to_excel(writer, sheet_name='价值型', index=False)
              momentum_picks.to_excel(writer, sheet_name='动量型', index=False)
              breakout_picks.to_excel(writer, sheet_name='突破型', index=False)
              multi_factor_picks.to_excel(writer, sheet_name='多因子', index=False)
          print(f"选股结果已保存至: {save_path}")
      
      return results
  
  
  if __name__ == "__main__":
      results = generate_daily_picks()
      
      print("\n" + "="*60)
      print(f"每日选股清单 - {results['date']}")
      print("="*60)
      
      print("\n【价值型选股 TOP10】")
      print(results['value'].head(10).to_string(index=False))
      
      print("\n【动量型选股 TOP10】")
      print(results['momentum'].head(10).to_string(index=False))
      
      print("\n【突破型选股 TOP10】")
      if len(results['breakout']) > 0:
          print(results['breakout'].head(10).to_string(index=False))
      else:
          print("今日无符合条件的突破型股票")
      
      print("\n【多因子选股 TOP10】")
      print(results['multi_factor'].head(10).to_string(index=False))
  

• lab/Geek-skills-a-share-analyst/scripts/technical_analysis.py11 KB8c198ca473d0upstream ↗

  #!/usr/bin/env python3
  """
  A股技术分析模块
  包含常用技术指标计算和信号判断
  """
  
  import pandas as pd
  import numpy as np
  
  
  def calc_ma(close: pd.Series, periods: list = [5, 10, 20, 60, 120]) -> pd.DataFrame:
      """计算移动平均线"""
      ma_df = pd.DataFrame(index=close.index)
      for p in periods:
          ma_df[f'MA{p}'] = close.rolling(window=p).mean()
      return ma_df
  
  
  def calc_ema(close: pd.Series, period: int = 12) -> pd.Series:
      """计算指数移动平均"""
      return close.ewm(span=period, adjust=False).mean()
  
  
  def calc_macd(close: pd.Series, fast: int = 12, slow: int = 26, signal: int = 9) -> dict:
      """计算MACD指标"""
      ema_fast = close.ewm(span=fast, adjust=False).mean()
      ema_slow = close.ewm(span=slow, adjust=False).mean()
      dif = ema_fast - ema_slow
      dea = dif.ewm(span=signal, adjust=False).mean()
      macd = (dif - dea) * 2
      
      return {
          'DIF': dif,
          'DEA': dea,
          'MACD': macd,
          'signal': get_macd_signal(dif, dea)
      }
  
  
  def get_macd_signal(dif: pd.Series, dea: pd.Series) -> str:
      """判断MACD信号"""
      if len(dif) < 2:
          return "数据不足"
      
      # 金叉/死叉判断
      if dif.iloc[-1] > dea.iloc[-1] and dif.iloc[-2] <= dea.iloc[-2]:
          return "金叉买入"
      elif dif.iloc[-1] < dea.iloc[-1] and dif.iloc[-2] >= dea.iloc[-2]:
          return "死叉卖出"
      elif dif.iloc[-1] > dea.iloc[-1]:
          if dif.iloc[-1] > 0:
              return "多头强势"
          else:
              return "多头初现"
      else:
          if dif.iloc[-1] < 0:
              return "空头强势"
          else:
              return "空头初现"
  
  
  def calc_kdj(high: pd.Series, low: pd.Series, close: pd.Series, 
               n: int = 9, m1: int = 3, m2: int = 3) -> dict:
      """计算KDJ指标"""
      low_n = low.rolling(window=n).min()
      high_n = high.rolling(window=n).max()
      rsv = (close - low_n) / (high_n - low_n) * 100
      rsv = rsv.fillna(50)
      
      k = rsv.ewm(com=m1-1, adjust=False).mean()
      d = k.ewm(com=m2-1, adjust=False).mean()
      j = 3 * k - 2 * d
      
      return {
          'K': k,
          'D': d,
          'J': j,
          'signal': get_kdj_signal(k, d)
      }
  
  
  def get_kdj_signal(k: pd.Series, d: pd.Series) -> str:
      """判断KDJ信号"""
      if len(k) < 2:
          return "数据不足"
      
      k_val = k.iloc[-1]
      
      if k.iloc[-1] > d.iloc[-1] and k.iloc[-2] <= d.iloc[-2]:
          if k_val < 20:
              return "超卖金叉-强买"
          elif k_val < 50:
              return "金叉买入"
          else:
              return "高位金叉"
      elif k.iloc[-1] < d.iloc[-1] and k.iloc[-2] >= d.iloc[-2]:
          if k_val > 80:
              return "超买死叉-强卖"
          elif k_val > 50:
              return "死叉卖出"
          else:
              return "低位死叉"
      elif k_val > 80:
          return "超买区域"
      elif k_val < 20:
          return "超卖区域"
      else:
          return "中性震荡"
  
  
  def calc_rsi(close: pd.Series, period: int = 14) -> dict:
      """计算RSI指标"""
      delta = close.diff()
      gain = delta.where(delta > 0, 0).rolling(window=period).mean()
      loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
      rs = gain / loss
      rsi = 100 - (100 / (1 + rs))
      
      return {
          'RSI': rsi,
          'signal': get_rsi_signal(rsi)
      }
  
  
  def get_rsi_signal(rsi: pd.Series) -> str:
      """判断RSI信号"""
      if len(rsi) < 1:
          return "数据不足"
      
      rsi_val = rsi.iloc[-1]
      
      if rsi_val > 80:
          return "严重超买"
      elif rsi_val > 70:
          return "超买"
      elif rsi_val < 20:
          return "严重超卖"
      elif rsi_val < 30:
          return "超卖"
      else:
          return f"中性({rsi_val:.1f})"
  
  
  def calc_boll(close: pd.Series, period: int = 20, std_dev: int = 2) -> dict:
      """计算布林带"""
      mid = close.rolling(window=period).mean()
      std = close.rolling(window=period).std()
      upper = mid + std_dev * std
      lower = mid - std_dev * std
      
      return {
          'UPPER': upper,
          'MID': mid,
          'LOWER': lower,
          'signal': get_boll_signal(close, upper, mid, lower)
      }
  
  
  def get_boll_signal(close: pd.Series, upper: pd.Series, mid: pd.Series, lower: pd.Series) -> str:
      """判断布林带信号"""
      if len(close) < 1:
          return "数据不足"
      
      price = close.iloc[-1]
      u = upper.iloc[-1]
      m = mid.iloc[-1]
      l = lower.iloc[-1]
      
      if price > u:
          return "突破上轨-超买"
      elif price < l:
          return "跌破下轨-超卖"
      elif price > m:
          return "中轨上方-偏多"
      else:
          return "中轨下方-偏空"
  
  
  def calc_atr(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 14) -> pd.Series:
      """计算ATR真实波幅"""
      tr1 = high - low
      tr2 = abs(high - close.shift(1))
      tr3 = abs(low - close.shift(1))
      tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
      return tr.rolling(window=period).mean()
  
  
  def calc_vol_ma(volume: pd.Series, periods: list = [5, 10, 20]) -> pd.DataFrame:
      """计算成交量均线"""
      vol_df = pd.DataFrame(index=volume.index)
      for p in periods:
          vol_df[f'VOL_MA{p}'] = volume.rolling(window=p).mean()
      return vol_df
  
  
  def detect_trend(close: pd.Series, ma_short: int = 20, ma_long: int = 60) -> str:
      """判断趋势方向"""
      ma_s = close.rolling(ma_short).mean()
      ma_l = close.rolling(ma_long).mean()
      
      if len(ma_s) < 1 or len(ma_l) < 1:
          return "数据不足"
      
      if ma_s.iloc[-1] > ma_l.iloc[-1] and close.iloc[-1] > ma_s.iloc[-1]:
          return "上升趋势"
      elif ma_s.iloc[-1] < ma_l.iloc[-1] and close.iloc[-1] < ma_s.iloc[-1]:
          return "下降趋势"
      else:
          return "震荡整理"
  
  
  def find_support_resistance(high: pd.Series, low: pd.Series, close: pd.Series, period: int = 20) -> dict:
      """计算支撑位和阻力位"""
      support = low.rolling(window=period).min().iloc[-1]
      resistance = high.rolling(window=period).max().iloc[-1]
      pivot = (high.iloc[-1] + low.iloc[-1] + close.iloc[-1]) / 3
      
      return {
          'support': round(support, 2),
          'resistance': round(resistance, 2),
          'pivot': round(pivot, 2)
      }
  
  
  def technical_score(macd_signal: str, kdj_signal: str, rsi_signal: str, 
                      boll_signal: str, trend: str) -> dict:
      """综合技术评分"""
      score = 50  # 基础分
      
      # MACD评分
      if '金叉' in macd_signal or '多头' in macd_signal:
          score += 15
      elif '死叉' in macd_signal or '空头' in macd_signal:
          score -= 15
      
      # KDJ评分
      if '强买' in kdj_signal:
          score += 15
      elif '买入' in kdj_signal:
          score += 10
      elif '强卖' in kdj_signal:
          score -= 15
      elif '卖出' in kdj_signal:
          score -= 10
      
      # RSI评分
      if '超卖' in rsi_signal:
          score += 10
      elif '超买' in rsi_signal:
          score -= 10
      
      # 布林带评分
      if '超卖' in boll_signal:
          score += 10
      elif '超买' in boll_signal:
          score -= 10
      elif '偏多' in boll_signal:
          score += 5
      elif '偏空' in boll_signal:
          score -= 5
      
      # 趋势评分
      if trend == "上升趋势":
          score += 15
      elif trend == "下降趋势":
          score -= 15
      
      # 技术面强弱描述（中性表述，非买卖评级——合规边界，见 SKILL.md）
      if score >= 80:
          rating = "偏多（强）"
          stars = 5
      elif score >= 65:
          rating = "偏多"
          stars = 4
      elif score >= 50:
          rating = "中性"
          stars = 3
      elif score >= 35:
          rating = "偏弱"
          stars = 2
      else:
          rating = "偏弱（强）"
          stars = 1
      
      return {
          'score': min(max(score, 0), 100),
          'rating': rating,
          'stars': stars
      }
  
  
  def analyze_stock(df: pd.DataFrame) -> dict:
      """
      完整技术分析
      
      Args:
          df: 包含 open, high, low, close, volume 的DataFrame
      """
      close = df['收盘'] if '收盘' in df.columns else df['close']
      high = df['最高'] if '最高' in df.columns else df['high']
      low = df['最低'] if '最低' in df.columns else df['low']
      volume = df['成交量'] if '成交量' in df.columns else df['volume']
      
      # 计算各项指标
      ma_data = calc_ma(close)
      macd_data = calc_macd(close)
      kdj_data = calc_kdj(high, low, close)
      rsi_data = calc_rsi(close)
      boll_data = calc_boll(close)
      atr = calc_atr(high, low, close)
      trend = detect_trend(close)
      sr = find_support_resistance(high, low, close)
      
      # 综合评分
      score_data = technical_score(
          macd_data['signal'],
          kdj_data['signal'],
          rsi_data['signal'],
          boll_data['signal'],
          trend
      )
      
      return {
          'price': {
              'current': round(close.iloc[-1], 2),
              'change_pct': round((close.iloc[-1] / close.iloc[-2] - 1) * 100, 2)
          },
          'trend': trend,
          'support_resistance': sr,
          'indicators': {
              'MACD': {
                  'DIF': round(macd_data['DIF'].iloc[-1], 4),
                  'DEA': round(macd_data['DEA'].iloc[-1], 4),
                  'signal': macd_data['signal']
              },
              'KDJ': {
                  'K': round(kdj_data['K'].iloc[-1], 2),
                  'D': round(kdj_data['D'].iloc[-1], 2),
                  'J': round(kdj_data['J'].iloc[-1], 2),
                  'signal': kdj_data['signal']
              },
              'RSI': {
                  'value': round(rsi_data['RSI'].iloc[-1], 2),
                  'signal': rsi_data['signal']
              },
              'BOLL': {
                  'upper': round(boll_data['UPPER'].iloc[-1], 2),
                  'mid': round(boll_data['MID'].iloc[-1], 2),
                  'lower': round(boll_data['LOWER'].iloc[-1], 2),
                  'signal': boll_data['signal']
              },
              'ATR': round(atr.iloc[-1], 2)
          },
          'score': score_data
      }
  
  
  if __name__ == "__main__":
      # 测试
      import akshare as ak
      
      df = ak.stock_zh_a_hist(symbol="000001", period="daily", adjust="qfq")
      df = df.tail(120)
      
      result = analyze_stock(df)
      print("=== 平安银行(000001) 技术分析 ===")
      print(f"当前价格: {result['price']['current']} ({result['price']['change_pct']:+.2f}%)")
      print(f"趋势判断: {result['trend']}")
      print(f"支撑位: {result['support_resistance']['support']} | 阻力位: {result['support_resistance']['resistance']}")
      print(f"\n技术指标:")
      for name, data in result['indicators'].items():
          print(f"  {name}: {data}")
      print(f"\n综合评分: {result['score']['score']} - {result['score']['rating']} ({'⭐' * result['score']['stars']})")
  

• lab/Geek-skills-a-share-analyst/references/candlestick_patterns.md3.9 KB87407babd6e6upstream ↗

  # K线形态识别
  
  ## 反转形态
  
  ### 底部反转
  
  **锤子线 (Hammer)**
  - 条件：下影线 >= 实体×2，上影线极小，出现在下跌末端
  ```python
  def is_hammer(open, high, low, close):
      body = abs(close - open)
      lower_shadow = min(open, close) - low
      upper_shadow = high - max(open, close)
      return lower_shadow >= body * 2 and upper_shadow <= body * 0.3
  ```
  
  **早晨之星 (Morning Star)**
  - 三根K线组合：大阴线 + 小实体(十字星) + 大阳线
  ```python
  def is_morning_star(df, idx):
      d1 = df.iloc[idx-2]  # 大阴线
      d2 = df.iloc[idx-1]  # 小实体
      d3 = df.iloc[idx]    # 大阳线
      return (d1['close'] < d1['open'] and 
              abs(d2['close'] - d2['open']) < (d1['high']-d1['low'])*0.3 and
              d3['close'] > d3['open'] and 
              d3['close'] > (d1['open'] + d1['close'])/2)
  ```
  
  **双底 (Double Bottom)**
  - W形态，两个低点接近，颈线突破确认
  
  ### 顶部反转
  
  **流星线 (Shooting Star)**
  - 条件：上影线 >= 实体×2，下影线极小，出现在上涨末端
  ```python
  def is_shooting_star(open, high, low, close):
      body = abs(close - open)
      upper_shadow = high - max(open, close)
      lower_shadow = min(open, close) - low
      return upper_shadow >= body * 2 and lower_shadow <= body * 0.3
  ```
  
  **黄昏之星 (Evening Star)**
  - 三根K线组合：大阳线 + 小实体(十字星) + 大阴线
  
  **双顶 (Double Top)**
  - M形态，两个高点接近，颈线跌破确认
  
  ## 持续形态
  
  ### 上升趋势中
  
  **红三兵 (Three White Soldiers)**
  - 三根连续阳线，每根开盘价在前一根实体内，收盘创新高
  ```python
  def is_three_white_soldiers(df, idx):
      for i in range(3):
          if df.iloc[idx-i]['close'] <= df.iloc[idx-i]['open']:
              return False
      return (df.iloc[idx]['close'] > df.iloc[idx-1]['close'] > df.iloc[idx-2]['close'])
  ```
  
  **上升三法 (Rising Three Methods)**
  - 大阳线 + 3根小阴线(不破大阳线低点) + 大阳线
  
  ### 下降趋势中
  
  **三只乌鸦 (Three Black Crows)**
  - 三根连续阴线，每根开盘价在前一根实体内，收盘创新低
  
  **下降三法 (Falling Three Methods)**
  - 大阴线 + 3根小阳线(不破大阴线高点) + 大阴线
  
  ## 整理形态
  
  ### 三角形整理
  
  **上升三角形**
  - 高点水平，低点上移，向上突破概率大
  ```python
  def is_ascending_triangle(highs, lows, period=20):
      high_max = highs[-period:].max()
      high_range = highs[-period:].max() - highs[-period:].min()
      low_slope = (lows.iloc[-1] - lows.iloc[-period]) / period
      return high_range / high_max < 0.02 and low_slope > 0
  ```
  
  **下降三角形**
  - 低点水平，高点下移，向下突破概率大
  
  **对称三角形**
  - 高点下移，低点上移，方向不确定
  
  ### 旗形整理
  
  **上升旗形**
  - 急涨后小幅回调整理，形成向下倾斜通道
  
  **下降旗形**
  - 急跌后小幅反弹整理，形成向上倾斜通道
  
  ## 缺口分析
  
  ### 缺口类型
  
  | 类型 | 特征 | 意义 |
  |------|------|------|
  | 普通缺口 | 整理区间内 | 通常会回补 |
  | 突破缺口 | 突破重要位置 | 趋势开始信号 |
  | 持续缺口 | 趋势中间 | 趋势加速信号 |
  | 衰竭缺口 | 趋势末端 | 趋势结束信号 |
  
  ```python
  def find_gaps(df):
      gaps = []
      for i in range(1, len(df)):
          if df.iloc[i]['low'] > df.iloc[i-1]['high']:  # 向上缺口
              gaps.append(('up', i, df.iloc[i-1]['high'], df.iloc[i]['low']))
          elif df.iloc[i]['high'] < df.iloc[i-1]['low']:  # 向下缺口
              gaps.append(('down', i, df.iloc[i]['high'], df.iloc[i-1]['low']))
      return gaps
  ```
  
  ## 实战信号组合
  
  ### 强买入信号
  1. 早晨之星 + MACD金叉 + 成交量放大
  2. 突破缺口 + 站上MA20 + KDJ金叉
  3. 锤子线 + RSI超卖反弹 + 布林下轨支撑
  
  ### 强卖出信号
  1. 黄昏之星 + MACD死叉 + 成交量放大
  2. 跌破颈线 + MA20死叉MA60 + KDJ死叉
  3. 流星线 + RSI超买回落 + 布林上轨压力
  

• lab/Geek-skills-a-share-analyst/references/factor_library.md6.7 KB1adcaef9ebd5upstream ↗

  # 量化因子库
  
  ## 价值因子
  
  ### EP (收益价格比)
  ```python
  def factor_ep(net_profit, market_cap):
      """EP = 净利润 / 市值，越高越便宜"""
      return net_profit / market_cap
  ```
  
  ### BP (账面市值比)
  ```python
  def factor_bp(book_value, market_cap):
      """BP = 净资产 / 市值，越高越便宜"""
      return book_value / market_cap
  ```
  
  ### SP (销售价格比)
  ```python
  def factor_sp(revenue, market_cap):
      """SP = 营收 / 市值"""
      return revenue / market_cap
  ```
  
  ### CFP (现金流价格比)
  ```python
  def factor_cfp(operating_cf, market_cap):
      """CFP = 经营现金流 / 市值"""
      return operating_cf / market_cap
  ```
  
  ### DP (股息率)
  ```python
  def factor_dp(dividend, price):
      """DP = 每股分红 / 股价"""
      return dividend / price
  ```
  
  ## 动量因子
  
  ### MOM (价格动量)
  ```python
  def factor_momentum(close, period=20):
      """过去N日收益率"""
      return close.pct_change(period)
  ```
  
  ### REV (反转因子)
  ```python
  def factor_reversal(close, period=5):
      """短期反转：过去N日收益率取负"""
      return -close.pct_change(period)
  ```
  
  ### VOL_MOM (成交量动量)
  ```python
  def factor_vol_momentum(volume, period=20):
      """成交量动量"""
      return volume.rolling(5).mean() / volume.rolling(period).mean()
  ```
  
  ### RSRS (阻力支撑相对强度)
  ```python
  def factor_rsrs(high, low, period=18):
      """回归斜率作为趋势强度"""
      from sklearn.linear_model import LinearRegression
      rsrs = []
      for i in range(period, len(high)):
          X = low[i-period:i].values.reshape(-1, 1)
          y = high[i-period:i].values
          reg = LinearRegression().fit(X, y)
          rsrs.append(reg.coef_[0])
      return pd.Series(rsrs, index=high.index[period:])
  ```
  
  ## 质量因子
  
  ### ROE (净资产收益率)
  ```python
  def factor_roe(net_profit, equity):
      return net_profit / equity
  ```
  
  ### ROIC (投入资本回报率)
  ```python
  def factor_roic(nopat, invested_capital):
      """ROIC = 税后经营利润 / 投入资本"""
      return nopat / invested_capital
  ```
  
  ### GP (毛利率)
  ```python
  def factor_gross_profit(revenue, cogs):
      return (revenue - cogs) / revenue
  ```
  
  ### ACCRUAL (应计盈余)
  ```python
  def factor_accrual(net_profit, operating_cf, total_assets):
      """应计盈余 = (净利润 - 经营现金流) / 总资产"""
      return (net_profit - operating_cf) / total_assets
  ```
  
  ## 规模因子
  
  ### SIZE (市值因子)
  ```python
  def factor_size(market_cap):
      """对数市值，通常取负表示小盘股溢价"""
      return -np.log(market_cap)
  ```
  
  ### NLSIZE (非线性市值)
  ```python
  def factor_nlsize(market_cap):
      """市值三次方根"""
      return np.cbrt(market_cap)
  ```
  
  ## 波动率因子
  
  ### VOL (波动率)
  ```python
  def factor_volatility(returns, period=20):
      """收益率标准差"""
      return returns.rolling(period).std()
  ```
  
  ### IVOL (特质波动率)
  ```python
  def factor_idio_vol(returns, market_returns, period=60):
      """剔除市场因素后的特质波动率"""
      from sklearn.linear_model import LinearRegression
      resid = []
      for i in range(period, len(returns)):
          X = market_returns[i-period:i].values.reshape(-1, 1)
          y = returns[i-period:i].values
          reg = LinearRegression().fit(X, y)
          pred = reg.predict(X)
          resid.append(np.std(y - pred))
      return pd.Series(resid, index=returns.index[period:])
  ```
  
  ### SKEW (偏度)
  ```python
  def factor_skewness(returns, period=20):
      return returns.rolling(period).skew()
  ```
  
  ## 技术因子
  
  ### TURN (换手率)
  ```python
  def factor_turnover(volume, float_shares, period=20):
      """平均换手率"""
      return (volume / float_shares).rolling(period).mean()
  ```
  
  ### ILLIQ (非流动性)
  ```python
  def factor_illiquidity(returns, volume, period=20):
      """Amihud非流动性指标"""
      return (returns.abs() / volume).rolling(period).mean()
  ```
  
  ### CORR (相关性)
  ```python
  def factor_correlation(returns, market_returns, period=60):
      """与市场的相关性"""
      return returns.rolling(period).corr(market_returns)
  ```
  
  ## 成长因子
  
  ### SUE (标准化盈余惊喜)
  ```python
  def factor_sue(actual_eps, expected_eps, std_eps):
      """SUE = (实际EPS - 预期EPS) / 标准差"""
      return (actual_eps - expected_eps) / std_eps
  ```
  
  ### REVG (营收增长率)
  ```python
  def factor_revenue_growth(revenue):
      return revenue.pct_change(4)  # 同比增长
  ```
  
  ### EPG (盈利增长率)
  ```python
  def factor_earnings_growth(net_profit):
      return net_profit.pct_change(4)  # 同比增长
  ```
  
  ## 因子合成与选股
  
  ### 多因子评分模型
  ```python
  def multi_factor_score(df, factors_config):
      """
      factors_config = {
          'EP': {'weight': 0.2, 'ascending': False},
          'MOM_20': {'weight': 0.15, 'ascending': False},
          'ROE': {'weight': 0.2, 'ascending': False},
          'REV_5': {'weight': 0.15, 'ascending': True},
          'VOL': {'weight': 0.1, 'ascending': True},
          'SIZE': {'weight': 0.2, 'ascending': True},
      }
      """
      scores = pd.DataFrame(index=df.index)
      for factor, config in factors_config.items():
          # 标准化因子值
          factor_values = df[factor]
          factor_rank = factor_values.rank(pct=True, ascending=config['ascending'])
          scores[factor] = factor_rank * config['weight']
      
      scores['total'] = scores.sum(axis=1)
      return scores.sort_values('total', ascending=False)
  ```
  
  ### 选股示例
  ```python
  def select_stocks(df, top_n=30):
      """
      选股流程：
      1. 剔除ST股票
      2. 剔除上市不满60天的新股
      3. 剔除涨停/跌停股票
      4. 计算多因子评分
      5. 选取评分前N名
      """
      # 基础筛选
      df = df[~df['name'].str.contains('ST')]
      df = df[df['list_days'] > 60]
      df = df[df['pct_change'].abs() < 9.9]
      
      # 计算因子评分
      scores = multi_factor_score(df, factors_config)
      
      return scores.head(top_n)
  ```
  
  ## 回测框架
  
  ```python
  def backtest_strategy(df, signal_col, hold_days=5):
      """
      简单回测框架
      df: 包含日期、收盘价、信号的DataFrame
      signal_col: 信号列名 (1=买入, 0=持有, -1=卖出)
      hold_days: 持仓天数
      """
      df['future_return'] = df['close'].shift(-hold_days) / df['close'] - 1
      df['strategy_return'] = df[signal_col].shift(1) * df['future_return']
      
      total_return = (1 + df['strategy_return']).cumprod().iloc[-1] - 1
      sharpe = df['strategy_return'].mean() / df['strategy_return'].std() * np.sqrt(252/hold_days)
      max_drawdown = (df['strategy_return'].cumsum() - df['strategy_return'].cumsum().cummax()).min()
      
      return {
          'total_return': total_return,
          'sharpe_ratio': sharpe,
          'max_drawdown': max_drawdown,
          'win_rate': (df['strategy_return'] > 0).mean()
      }
  ```
  

• lab/Geek-skills-a-share-analyst/references/fundamental_metrics.md4.0 KB6dc169e19b55upstream ↗

  # 基本面指标说明
  
  ## 估值指标
  
  ### PE (市盈率)
  ```
  PE = 股价 / 每股收益(EPS)
  PE(TTM) = 当前市值 / 最近四个季度净利润
  ```
  - PE < 15：低估值
  - 15 < PE < 25：合理估值
  - PE > 25：高估值
  - 注意：亏损股票PE无意义，需结合行业对比
  
  ### PB (市净率)
  ```
  PB = 股价 / 每股净资产
  ```
  - PB < 1：破净股，可能低估或资产质量差
  - 1 < PB < 3：合理区间
  - PB > 5：可能高估
  
  ### PS (市销率)
  ```
  PS = 当前市值 / 年度营业收入
  ```
  - 适用于亏损但有收入增长的公司
  - PS < 1：可能低估
  - PS > 10：可能高估
  
  ### PEG (市盈增长比率)
  ```
  PEG = PE / 净利润增长率(%)
  ```
  - PEG < 1：相对低估
  - PEG = 1：合理
  - PEG > 1：相对高估
  
  ## 盈利能力
  
  ### ROE (净资产收益率)
  ```
  ROE = 净利润 / 平均股东权益 × 100%
  ```
  - ROE > 15%：优秀
  - 10% < ROE < 15%：良好
  - ROE < 10%：一般
  
  ### ROA (总资产收益率)
  ```
  ROA = 净利润 / 平均总资产 × 100%
  ```
  - ROA > 10%：优秀
  - 5% < ROA < 10%：良好
  
  ### 毛利率
  ```
  毛利率 = (营业收入 - 营业成本) / 营业收入 × 100%
  ```
  - 毛利率 > 40%：高毛利行业(医药、软件)
  - 20% < 毛利率 < 40%：中等水平
  - 毛利率 < 20%：低毛利行业(零售、制造)
  
  ### 净利率
  ```
  净利率 = 净利润 / 营业收入 × 100%
  ```
  - 净利率 > 15%：优秀
  - 5% < 净利率 < 15%：良好
  
  ## 成长能力
  
  ### 营收增长率
  ```
  营收增长率 = (本期营收 - 上期营收) / 上期营收 × 100%
  ```
  - 增速 > 30%：高成长
  - 10% < 增速 < 30%：中等成长
  - 增速 < 10%：稳定型
  
  ### 净利润增长率
  ```
  净利增长率 = (本期净利 - 上期净利) / 上期净利 × 100%
  ```
  - 持续稳定增长优于大起大落
  
  ### 扣非净利润增长率
  - 排除非经常性损益的净利润增长
  - 更能反映主营业务真实增长
  
  ## 财务健康
  
  ### 资产负债率
  ```
  资产负债率 = 总负债 / 总资产 × 100%
  ```
  - < 40%：财务稳健
  - 40% - 60%：正常水平
  - > 70%：财务风险较高
  - 注意：银行等金融行业除外
  
  ### 流动比率
  ```
  流动比率 = 流动资产 / 流动负债
  ```
  - > 2：短期偿债能力强
  - 1 - 2：正常
  - < 1：短期偿债压力大
  
  ### 速动比率
  ```
  速动比率 = (流动资产 - 存货) / 流动负债
  ```
  - > 1：偿债能力良好
  
  ### 经营现金流/净利润
  ```
  现金流比率 = 经营活动现金流净额 / 净利润
  ```
  - > 1：利润质量高，现金回收好
  - < 0.5：利润质量存疑
  
  ## 运营效率
  
  ### 存货周转率
  ```
  存货周转率 = 营业成本 / 平均存货
  存货周转天数 = 365 / 存货周转率
  ```
  - 周转率越高，运营效率越好
  
  ### 应收账款周转率
  ```
  应收周转率 = 营业收入 / 平均应收账款
  应收周转天数 = 365 / 应收周转率
  ```
  - 周转天数短：回款快，坏账风险低
  
  ### 总资产周转率
  ```
  总资产周转率 = 营业收入 / 平均总资产
  ```
  - 反映资产利用效率
  
  ## 分红指标
  
  ### 股息率
  ```
  股息率 = 每股分红 / 股价 × 100%
  ```
  - 股息率 > 3%：高股息股票
  - 长期稳定分红表明财务健康
  
  ### 分红率
  ```
  分红率 = 每股分红 / 每股收益 × 100%
  ```
  - 合理分红率：30%-50%
  
  ## AKShare获取基本面数据
  
  ```python
  import akshare as ak
  
  # 财务指标
  df = ak.stock_financial_analysis_indicator(symbol="000001")
  
  # 资产负债表
  df = ak.stock_balance_sheet_by_report_em(symbol="000001")
  
  # 利润表
  df = ak.stock_profit_sheet_by_report_em(symbol="000001")
  
  # 现金流量表
  df = ak.stock_cash_flow_sheet_by_report_em(symbol="000001")
  
  # 业绩预告
  df = ak.stock_yjyg_em()
  
  # 业绩快报
  df = ak.stock_yjkb_em()
  ```
  
  ## 综合评分模型
  
  | 维度 | 指标 | 权重 | 优秀 | 良好 | 一般 |
  |------|------|------|------|------|------|
  | 估值 | PE | 20% | <15 | 15-25 | >25 |
  | 估值 | PB | 10% | <2 | 2-4 | >4 |
  | 盈利 | ROE | 20% | >15% | 10-15% | <10% |
  | 盈利 | 净利率 | 10% | >15% | 5-15% | <5% |
  | 成长 | 营收增速 | 15% | >30% | 10-30% | <10% |
  | 成长 | 净利增速 | 15% | >30% | 10-30% | <10% |
  | 安全 | 资产负债率 | 10% | <40% | 40-60% | >60% |
  

• lab/Geek-skills-a-share-analyst/references/technical_indicators.md4.0 KB4937eff9994bupstream ↗

  # 技术指标计算公式
  
  ## 趋势指标
  
  ### MA (移动平均线)
  ```python
  def calc_ma(close, period):
      return close.rolling(window=period).mean()
  # 常用周期：MA5, MA10, MA20, MA60, MA120, MA250
  ```
  
  ### EMA (指数移动平均)
  ```python
  def calc_ema(close, period):
      return close.ewm(span=period, adjust=False).mean()
  ```
  
  ### MACD (指数平滑异同移动平均线)
  ```python
  def calc_macd(close, fast=12, slow=26, signal=9):
      ema_fast = close.ewm(span=fast, adjust=False).mean()
      ema_slow = close.ewm(span=slow, adjust=False).mean()
      dif = ema_fast - ema_slow  # 快线
      dea = dif.ewm(span=signal, adjust=False).mean()  # 慢线
      macd = (dif - dea) * 2  # 柱状图
      return dif, dea, macd
  # 金叉：DIF上穿DEA | 死叉：DIF下穿DEA
  ```
  
  ## 震荡指标
  
  ### KDJ (随机指标)
  ```python
  def calc_kdj(high, low, close, n=9, m1=3, m2=3):
      low_n = low.rolling(window=n).min()
      high_n = high.rolling(window=n).max()
      rsv = (close - low_n) / (high_n - low_n) * 100
      k = rsv.ewm(com=m1-1, adjust=False).mean()
      d = k.ewm(com=m2-1, adjust=False).mean()
      j = 3 * k - 2 * d
      return k, d, j
  # K>80超买 | K<20超卖 | 金叉：K上穿D
  ```
  
  ### RSI (相对强弱指标)
  ```python
  def calc_rsi(close, period=14):
      delta = close.diff()
      gain = delta.where(delta > 0, 0).rolling(window=period).mean()
      loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
      rs = gain / loss
      return 100 - (100 / (1 + rs))
  # RSI>70超买 | RSI<30超卖
  ```
  
  ### WR (威廉指标)
  ```python
  def calc_wr(high, low, close, period=14):
      hh = high.rolling(window=period).max()
      ll = low.rolling(window=period).min()
      return -100 * (hh - close) / (hh - ll)
  # WR>-20超买 | WR<-80超卖
  ```
  
  ## 波动率指标
  
  ### BOLL (布林带)
  ```python
  def calc_boll(close, period=20, std_dev=2):
      mid = close.rolling(window=period).mean()
      std = close.rolling(window=period).std()
      upper = mid + std_dev * std
      lower = mid - std_dev * std
      return upper, mid, lower
  # 突破上轨：超买/强势 | 跌破下轨：超卖/弱势
  ```
  
  ### ATR (真实波动幅度)
  ```python
  def calc_atr(high, low, close, period=14):
      tr1 = high - low
      tr2 = abs(high - close.shift(1))
      tr3 = abs(low - close.shift(1))
      tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
      return tr.rolling(window=period).mean()
  # 用于止损设置：止损位 = 入场价 - N * ATR
  ```
  
  ## 成交量指标
  
  ### OBV (能量潮)
  ```python
  def calc_obv(close, volume):
      direction = np.where(close > close.shift(1), 1, 
                  np.where(close < close.shift(1), -1, 0))
      return (direction * volume).cumsum()
  # OBV上升趋势配合价格上涨确认趋势
  ```
  
  ### VOL_MA (成交量均线)
  ```python
  def calc_vol_ma(volume, period):
      return volume.rolling(window=period).mean()
  # 放量突破：当日成交量 > 2 * VOL_MA5
  ```
  
  ## 形态识别
  
  ### 趋势判断
  ```python
  def trend_direction(close, ma_short=20, ma_long=60):
      ma_s = close.rolling(ma_short).mean()
      ma_l = close.rolling(ma_long).mean()
      if ma_s.iloc[-1] > ma_l.iloc[-1] and close.iloc[-1] > ma_s.iloc[-1]:
          return "上升趋势"
      elif ma_s.iloc[-1] < ma_l.iloc[-1] and close.iloc[-1] < ma_s.iloc[-1]:
          return "下降趋势"
      else:
          return "震荡整理"
  ```
  
  ### 支撑阻力位
  ```python
  def support_resistance(high, low, close, period=20):
      support = low.rolling(window=period).min()
      resistance = high.rolling(window=period).max()
      pivot = (high + low + close) / 3
      return support.iloc[-1], resistance.iloc[-1], pivot.iloc[-1]
  ```
  
  ## 信号评分系统
  
  综合评分 = 趋势得分×0.3 + 动量得分×0.3 + 量能得分×0.2 + 波动得分×0.2
  
  | 指标 | 强买入 | 买入 | 中性 | 卖出 | 强卖出 |
  |------|--------|------|------|------|--------|
  | MACD | DIF>0且金叉 | 金叉 | 中性 | 死叉 | DIF<0且死叉 |
  | KDJ | K<20金叉 | K<50金叉 | 40<K<60 | K>50死叉 | K>80死叉 |
  | RSI | RSI<30 | 30<RSI<40 | 40<RSI<60 | 60<RSI<70 | RSI>70 |
  | BOLL | 下轨反弹 | 中轨支撑 | 中轨附近 | 中轨压力 | 上轨回落 |
  

• lab/Geek-skills-a-share-analyst/evals/routing-evals.json1.6 KB7ac7680f808cupstream ↗

  [
    {
      "id": "ash-01",
      "prompt": "帮我分析一下这只 A 股的技术面，MACD、KDJ、支撑阻力都看看。",
      "should_trigger": true,
      "reason": "A-share technical analysis"
    },
    {
      "id": "ash-02",
      "prompt": "今天大盘怎么样，帮我筛几只符合多因子策略的票。",
      "should_trigger": true,
      "reason": "market overview + factor screening"
    },
    {
      "id": "ash-03",
      "prompt": "这家公司的估值和财务指标帮我看看贵不贵。",
      "should_trigger": true,
      "reason": "场景触发：基本面/估值分析，未出现'股票'关键词"
    },
    {
      "id": "ash-04",
      "prompt": "帮我调研新能源汽车行业的政策和市场前景，要有引用。",
      "should_trigger": false,
      "route_to": "deep-research",
      "reason": "cited industry research, not stock TA/FA"
    },
    {
      "id": "ash-05",
      "prompt": "这只美股 TSLA 帮我分析一下。",
      "should_trigger": false,
      "route_to": "none",
      "reason": "US equity; data source and rules differ, out of scope"
    },
    {
      "id": "ash-06",
      "prompt": "帮我预测一下明天这只票会涨还是跌，我该不该满仓。",
      "should_trigger": false,
      "route_to": "none",
      "reason": "asks for a directional prediction / buy order — explicitly outside the compliance boundary"
    },
    {
      "id": "ash-07",
      "prompt": "帮我把这只票的分析结论做成一份可发的图文报告。",
      "should_trigger": true,
      "boundary": true,
      "reason": "两可：分析主体归本 skill，但'做成可发图文'的视觉化归 deck-studio；本 skill 出分析内容，传播图另行移交"
    }
  ]
  

• README.mdupstream ↗

  Contents are not inlined in the catalog. Open the upstream file.

Plain-text copy, no login: /skill-bundles/staruhub-claudeskills-lab-geek-skills-a-share-analyst.md