1. 项目概述：当AI遇见投资决策

最近在折腾一个挺有意思的东西，叫OpenClaw。这名字听起来有点“硬核”，但说白了，它就是一个帮你把大语言模型（比如GPT-4、Claude、国内的各种大模型）的能力，封装成一个个可执行、可组合“技能”的框架。你可以把它想象成一个“乐高积木”平台，每个积木（Skill）都是一个特定的功能模块，比如读取新闻、分析财报、计算技术指标。而Agent，就是那个按照你设定的逻辑，去调用这些积木完成复杂任务的“总工程师”。

那么，“用OpenClaw打造AI选股系统”这个标题意味着什么？它绝不是指开发一个能百分百预测股价、让你一夜暴富的“圣杯”。任何这样承诺的系统都是不切实际的。它的核心价值在于，将投资研究中那些重复、繁琐、基于规则的信息处理工作自动化，从而把你——投资者——从海量数据的泥潭中解放出来，让你能更专注于最终的策略思考和决策。

举个例子，你每天可能要手动查看几十份公司公告、追踪多个宏观经济指标、计算一系列财务比率。这个过程耗时耗力，还容易出错。一个基于OpenClaw构建的AI选股系统，可以自动完成这些工作：定时抓取数据、按你的规则进行初步筛选、生成结构化的分析报告，甚至给出基于历史模式的概率性提示。它充当的是一个不知疲倦、高度定制化的初级研究员角色。

这个项目适合谁呢？首先是对Python有一定了解，愿意动手的技术爱好者或量化交易新手。其次，是那些对现有炒股软件提供的标准化选股器感到不满，希望拥有更灵活、更贴合个人投资逻辑的工具的投资者。最后，它也适合想了解AI Agent如何在实际业务场景中落地的开发者。整个过程，从环境搭建到跑通第一个选股策略，顺利的话，核心部分确实能在10分钟左右搞定，但后续的调优和深化，才是真正体现价值的地方。

2. 核心思路：拆解一个AI选股系统的构成

要构建这个系统，我们得先抛开“AI”这个有点玄乎的光环，回归到选股这件事的本质。一个典型的、个人投资者层面的选股流程，通常包含几个环节：数据获取 -> 指标计算 -> 规则筛选 -> 结果呈现。OpenClaw的用武之地，就是把这些环节模块化、自动化。

2.1 理解OpenClaw的核心概念：Skill与Agent

这是理解整个项目的基石。OpenClaw的架构非常清晰：

• Skill（技能）：这是最小的功能单元。一个Skill只做一件事，并且把它做好。比如：

  • fetch_financial_data_skill: 从某个数据源（如新浪财经、雅虎财经的公开接口）获取某只股票的历史价格或财务报表。
  • calculate_ma_skill: 计算移动平均线。
  • screening_pe_skill: 筛选市盈率低于某个阈值的股票。
  • generate_report_skill: 将筛选结果整理成一份Markdown或文本报告。 每个Skill都有明确的输入和输出。在OpenClaw中，Skill通常用Python函数来实现，并通过装饰器进行注册。

• Agent（智能体）：这是负责编排和决策的“大脑”。Agent定义了工作的流程和逻辑。它来决定先调用哪个Skill，后调用哪个Skill，以及如何处理中间结果。在我们的选股系统里，Agent就是整个选股策略的执行者。一个简单的Agent工作流可能是：1. 调用Skill获取股票池列表；2. 遍历股票池，对每只股票调用财务数据获取和指标计算Skill；3. 调用筛选Skill过滤股票；4. 调用报告生成Skill输出结果。

• 大模型集成：OpenClaw的强大之处在于，它可以无缝集成大语言模型。这意味着，你的Skill或Agent可以利用大模型的推理和自然语言处理能力。例如，你可以开发一个news_sentiment_analysis_skill，它调用大模型API，对抓取到的公司新闻进行情感分析，输出“积极”、“消极”或“中性”的标签及置信度，作为选股的一个因子。

所以，构建选股系统的过程，就变成了：将你的选股逻辑分解成多个独立的Skill，然后设计一个Agent来把它们串联起来。

2.2 选股策略的模块化设计

我们以一个非常经典（但过于简单，仅作示例）的“双低”策略为例：寻找市盈率（PE）低且市净率（PB）低的股票。我们可以这样设计Skill：

1. 数据层Skill：

   • get_stock_list_skill: 获取全市场或某个板块的股票代码列表。
   • fetch_latest_financial_skill: 根据股票代码，获取其最新的PE和PB数据。这里的数据可以来自免费的财经API（如akshare、yfinance），也可以是你自己维护的数据库。

2. 计算/处理层Skill：

   • 在这个简单例子中，数据获取后可能无需复杂计算。但如果策略更复杂，比如需要计算ROE（净资产收益率）= 净利润 / 净资产，那么就需要一个calculate_roe_skill。

3. 筛选层Skill：

   • screen_low_pe_pb_skill: 接收一个包含股票代码、PE、PB的数据列表，按照预设的阈值（例如PE<20且PB<2）进行过滤，返回符合条件的股票列表。

4. 输出层Skill：

   • output_to_csv_skill: 将最终结果保存为CSV文件。
   • generate_summary_skill: 调用大模型，对筛选出的股票列表进行简要总结，例如“本次筛选出5只股票，主要集中在银行和地产板块，平均PE为15，平均PB为1.5”。

5. 编排层（Agent）：

   • 创建一个low_pe_pb_agent，它按顺序执行：获取股票列表 -> 批量获取财务数据 -> 执行双低筛选 -> 输出结果到文件并生成文本摘要。

注意：在真实投资中，“双低”策略存在明显的局限性（如陷入“价值陷阱”），这里仅用于演示技术实现路径。一个严肃的策略需要考虑更多维度，如行业景气度、公司成长性、财务质量等。

2.3 技术栈与工具选型

为什么选择OpenClaw而不是自己从头写脚本？因为OpenClaw提供了标准化的Skill定义、生命周期管理和Agent编排框架，让系统更容易维护、扩展和复用。你的技术栈将围绕它展开：

• 核心框架：OpenClaw。这是项目的基石。
• 编程语言：Python。这是OpenClaw和绝大多数数据科学、AI库的首选语言。
• 数据获取：
  • akshare：一个非常好用的中文财经数据接口库，免费、全面，覆盖A股、港股、美股、宏观经济等。
  • yfinance：获取美股、全球主要市场数据的经典库。
  • Tushare或Baostock：国内老牌的金融数据平台，部分高级数据需要积分或付费。
  • 自己搭建的数据管道：如果需要更稳定、更定制化的数据，可以考虑用爬虫（配合Scrapy、Selenium）或购买商业数据API，然后将数据清洗后存入数据库（如MySQL、PostgreSQL）。
• 数据处理与分析：pandas和numpy。这是处理表格数据和数值计算的黄金组合。
• 大模型接入：
  • OpenAI API(GPT系列)：全球最主流，但需要处理网络访问和付费。
  • 国内大模型API：如智谱AI（ChatGLM）、百度文心一言、阿里通义千问、月之暗面（Kimi）等。这些对于国内开发者来说访问更稳定。
  • 本地大模型：通过Ollama、LM Studio或vLLM等工具在本地部署开源模型（如Qwen、Llama系列、ChatGLM3）。这提供了完全的隐私和控制权，但对硬件有要求。
• 部署与调度：
  • Docker：将整个OpenClaw项目及其依赖打包成容器，实现环境一致性，方便在任何地方部署。
  • 任务调度：如果需要每天定时运行选股策略，可以使用crontab（Linux/Mac）或Task Scheduler（Windows），更复杂的可以用Apache Airflow或Celery。

3. 十分钟快速上手：搭建你的第一个选股Skill

理论说了这么多，我们直接动手，在10分钟内创建一个最简单的、可运行的选股Skill。假设你已经安装了Python（3.8以上）和pip。

3.1 环境准备与OpenClaw安装

首先，创建一个干净的虚拟环境是个好习惯，可以避免包冲突。

# 创建并激活虚拟环境（以conda为例，也可以用venv） conda create -n openclaw-stock python=3.10 conda activate openclaw-stock # 安装OpenClaw。请注意，OpenClaw可能仍在快速迭代，最稳妥的方式是从其官方GitHub仓库安装。 # 通常使用pip从git直接安装 pip install "openclaw @ git+https://github.com/openclaw-ai/openclaw.git" # 如果上述命令不行，可以尝试克隆仓库后本地安装 # git clone https://github.com/openclaw-ai/openclaw.git # cd openclaw # pip install -e .

安装完成后，验证一下是否成功。可以尝试在Python中导入：import openclaw。如果没有报错，说明基础框架安装好了。

接下来安装我们需要的财经数据库。这里我们使用akshare，因为它对中文市场支持非常好且免费。

pip install akshare pandas

3.2 编写第一个数据获取Skill

我们来创建一个获取A股股票基本信息的Skill。在项目目录下，新建一个Python文件，比如stock_skills.py。

# stock_skills.py import akshare as ak import pandas as pd from typing import List, Dict, Any from openclaw import skill @skill( name="get_a_stock_list_skill", description="获取全A股股票代码和名称列表", inputs={}, # 这个Skill不需要输入参数 outputs={"stock_list": List[Dict[str, Any]]} # 输出一个字典列表 ) def get_a_stock_list() -> Dict[str, Any]: """ 使用akshare获取沪深京A股列表 """ try: # 调用akshare接口 stock_info_a_code_name_df = ak.stock_info_a_code_name() # 将DataFrame转换为字典列表，方便后续处理 # 假设返回的DataFrame包含‘code’和‘name’两列 stock_list = stock_info_a_code_name_df[['code', 'name']].to_dict('records') return {"stock_list": stock_list} except Exception as e: # 良好的错误处理很重要 return {"stock_list": [], "error": f"获取股票列表失败: {str(e)}"} # 我们可以再写一个获取股票估值的Skill，展示如何接收输入 @skill( name="get_stock_valuation_skill", description="根据股票代码获取其最新估值数据（PE, PB）", inputs={"stock_code": str}, # 输入一个股票代码 outputs={"valuation": Dict[str, Any]} ) def get_stock_valuation(stock_code: str) -> Dict[str, Any]: """ 获取单只股票的市盈率和市净率。 注意：akshare的估值接口可能有时效性，且不同接口数据源不同，这里使用`stock_a_indicator`接口示例。 实际应用中需要确认接口稳定性和数据准确性。 """ try: # 这里以获取个股指标为例，该接口可能包含PE、PB等数据 indicator_df = ak.stock_a_indicator_lg(symbol=stock_code) if not indicator_df.empty: latest = indicator_df.iloc[-1] # 取最新一行数据 pe = latest.get('pe_ttm', None) # 滚动市盈率 pb = latest.get('pb', None) # 市净率 return { "valuation": { "stock_code": stock_code, "pe": pe, "pb": pb, "date": latest.name if hasattr(latest, 'name') else None } } else: return {"valuation": {"stock_code": stock_code, "error": "未找到数据"}} except Exception as e: return {"valuation": {"stock_code": stock_code, "error": str(e)}}

这两个Skill已经具备了基本功能。第一个无参数，返回全市场股票列表。第二个需要输入股票代码，返回该股票的估值数据。我们用@skill装饰器来声明它们，这样OpenClaw就能识别和管理这些技能。

3.3 创建并运行一个简单的选股Agent

现在，我们创建一个Agent来使用这些Skill。新建一个文件simple_screening_agent.py。

# simple_screening_agent.py import asyncio from openclaw import Agent from stock_skills import get_a_stock_list, get_stock_valuation # 定义一个简单的选股Agent class SimpleScreeningAgent(Agent): name = "simple_screening_agent" description = "一个简单的低PE筛选Agent" async def run(self): """ Agent的执行逻辑 """ self.logger.info("开始执行简单选股策略...") # 1. 获取股票列表 stock_list_result = await get_a_stock_list() all_stocks = stock_list_result.get("stock_list", []) self.logger.info(f"共获取到 {len(all_stocks)} 只股票") if not all_stocks: self.logger.error("未获取到股票列表，任务终止。") return {"result": []} # 2. 遍历股票列表，获取估值（这里简单起见，只处理前10只，避免请求过多） screened_stocks = [] sample_stocks = all_stocks[:10] # 仅示例，处理前10只 for stock in sample_stocks: stock_code = stock.get('code') if not stock_code: continue self.logger.debug(f"正在处理股票: {stock_code}") valuation_result = await get_stock_valuation(stock_code) valuation_data = valuation_result.get("valuation", {}) # 3. 简单的筛选逻辑：PE存在且小于30 pe = valuation_data.get('pe') if pe is not None and pe < 30 and pe > 0: # 注意PE可能为负或无穷大 screened_stocks.append({ "code": stock_code, "name": stock.get('name'), "pe": pe, "pb": valuation_data.get('pb') }) # 4. 输出结果 self.logger.info(f"筛选完成，共找到 {len(screened_stocks)} 只符合条件的股票") for s in screened_stocks: self.logger.info(f" - {s['code']} ({s['name']}): PE={s['pe']:.2f}, PB={s.get('pb', 'N/A')}") return {"screened_stocks": screened_stocks} # 运行这个Agent if __name__ == "__main__": agent = SimpleScreeningAgent() result = asyncio.run(agent.run()) print("最终结果:", result)

这个Agent做了四件事：获取列表、遍历获取估值、应用筛选规则（PE<30）、输出结果。现在，在终端运行这个文件：

python simple_screening_agent.py

如果一切顺利，你将在控制台看到日志输出，以及筛选出的股票列表。恭喜你，你的第一个AI选股系统（雏形）已经跑起来了！从安装到运行，核心步骤确实可以在十分钟内完成。

实操心得：第一次运行很可能遇到问题。最常见的是akshare接口变更或网络问题。不要慌，这是常态。解决方法是：1. 查看akshare的官方文档或GitHub Issues，确认接口用法；2. 在Skill函数内部添加更详细的异常打印和日志；3. 对于网络问题，可以考虑添加重试机制或使用代理（此处指HTTP代理，用于网络请求，与系统网络配置相关）。数据源的稳定性是这类系统最大的挑战之一。

4. 系统深化：构建一个实用的选股工作流

上面的例子只是一个演示。一个实用的系统需要更健壮、更高效、更灵活。我们来深入构建一个更接近实际应用的版本。

4.1 设计一个完整的选股Agent工作流

一个完整的工作流应该考虑并发、错误处理、结果持久化和可配置性。我们设计一个名为ValueScreeningAgent的Agent，它包含以下步骤：

1. 配置加载：从配置文件（如YAML）中读取筛选参数（PE上限、PB上限、行业排除列表等）。
2. 股票池获取：调用Skill获取初始股票池。可以不止一个来源，比如合并主板和创业板的股票。
3. 并发数据获取：同时为多只股票获取财务和估值数据，大幅提升效率。OpenClaw的Agent可以利用asyncio.gather来实现并发。
4. 多因子筛选：应用多个筛选条件，可能包括：估值因子（PE, PB）、质量因子（ROE, 毛利率）、增长因子（营收增长率）、市场因子（市值）。每个因子可以是一个独立的Skill。
5. 综合排序：对通过筛选的股票，根据一个综合评分模型进行排序（例如，PE分位点越低得分越高，ROE越高得分越高，加权求和）。
6. 报告生成：将最终结果（如前20名股票）生成一份详细的报告，包括列表、关键指标、甚至简单的文本点评（调用大模型Skill）。
7. 结果推送：将报告通过邮件、飞书/钉钉机器人、或者保存到数据库/指定文件路径。

4.2 关键Skill的实现与优化

让我们实现几个更健壮的Skill。

（1）带缓存和重试的数据获取Skill

网络请求不稳定，频繁请求也可能被数据源限制。我们需要一个更强大的数据获取Skill。

# advanced_data_skill.py import akshare as ak import pandas as pd import asyncio import aiohttp import json import time from typing import Optional from openclaw import skill from functools import lru_cache import hashlib # 简单的内存缓存装饰器（对于短期重复请求有效） def cache_ttl(ttl_seconds=300): def decorator(func): cache = {} def wrapper(*args, **kwargs): # 创建一个基于参数的唯一缓存键 key = hashlib.md5(json.dumps((args, sorted(kwargs.items())), default=str).encode()).hexdigest() if key in cache: timestamp, value = cache[key] if time.time() - timestamp < ttl_seconds: return value result = func(*args, **kwargs) cache[key] = (time.time(), result) return result return wrapper return decorator @skill( name="get_stock_daily_skill", description="获取股票日线行情数据，带缓存和重试", inputs={"stock_code": str, "start_date": Optional[str], "end_date": Optional[str]}, outputs={"daily_data": list} ) @cache_ttl(ttl_seconds=600) # 缓存10分钟 async def get_stock_daily(stock_code: str, start_date: str = None, end_date: str = None) -> dict: """ 使用akshare获取日线数据，加入异步和重试逻辑。 """ max_retries = 3 for i in range(max_retries): try: # akshare目前主要是同步接口，可以放在线程池中运行避免阻塞 loop = asyncio.get_event_loop() # 注意：ak.stock_zh_a_hist 是同步函数，用run_in_executor防止阻塞事件循环 df = await loop.run_in_executor(None, ak.stock_zh_a_hist, stock_code, "daily", start_date, end_date) if df is not None and not df.empty: return {"daily_data": df.to_dict('records')} else: return {"daily_data": [], "error": "数据为空"} except Exception as e: if i == max_retries - 1: return {"daily_data": [], "error": f"获取{stock_code}日线数据失败: {str(e)}"} await asyncio.sleep(2 ** i) # 指数退避 return {"daily_data": [], "error": "重试次数用尽"}

（2）技术指标计算Skill

有了行情数据，我们可以计算技术指标。

# technical_skill.py import pandas as pd import numpy as np from openclaw import skill @skill( name="calculate_technical_indicators_skill", description="计算一系列技术指标（MA, RSI, MACD等）", inputs={"daily_data": list, "indicators": list}, # indicators可以是['ma20', 'rsi14', 'macd'] outputs={"indicators_data": dict} ) def calculate_technical_indicators(daily_data: list, indicators: list) -> dict: """ 基于日线数据计算技术指标。 daily_data 是一个列表，每个元素是包含‘date’, ‘close’, ‘high’, ‘low’, ‘volume’等的字典。 """ if not daily_data: return {"indicators_data": {}} df = pd.DataFrame(daily_data) # 确保数据按日期排序 if 'date' in df.columns: df['date'] = pd.to_datetime(df['date']) df = df.sort_values('date') result = {} close_series = df['close'] for indicator in indicators: if indicator.startswith('ma'): try: period = int(indicator[2:]) # 如‘ma20’ -> 20 result[f'ma{period}'] = close_series.rolling(window=period).mean().iloc[-1] except: pass elif indicator == 'rsi14': # 简化版RSI计算 delta = close_series.diff() gain = (delta.where(delta > 0, 0)).rolling(window=14).mean() loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean() rs = gain / loss rsi = 100 - (100 / (1 + rs)) result['rsi14'] = rsi.iloc[-1] if not pd.isna(rsi.iloc[-1]) else None return {"indicators_data": result}

（3）集成大模型的报告生成Skill

这是体现“AI”价值的关键一环。我们让大模型来为我们写一段简单的分析。

# llm_report_skill.py import openai # 或 from openai import OpenAI # 或者使用国内模型的SDK，如 from zhipuai import ZhipuAI from openclaw import skill import json @skill( name="generate_stock_comment_skill", description="使用大模型为筛选出的股票列表生成简要评论", inputs={"screened_stocks": list, "strategy_description": str}, outputs={"comment": str} ) async def generate_stock_comment(screened_stocks: list, strategy_description: str) -> dict: """ 调用大模型API，生成一段关于筛选结果的文本评论。 """ if not screened_stocks: return {"comment": "本次筛选未找到符合条件的股票。"} # 1. 准备Prompt stocks_info = "" for s in screened_stocks[:10]: # 限制前10只，避免token过长 stocks_info += f"- 代码：{s.get('code')}， 名称：{s.get('name')}， PE：{s.get('pe')}， PB：{s.get('pb')}， 行业：{s.get('industry', 'N/A')}\\n" prompt = f""" 你是一个专业的股票分析师。以下是根据“{strategy_description}”策略筛选出的一批股票列表及其关键数据： {stocks_info} 请用一段话（不超过200字）简要总结这批股票的整体特征，例如：它们主要集中在哪些行业？估值水平整体如何？有哪些潜在的风险点需要关注？请保持分析中立、客观。 """ # 2. 调用大模型API（以OpenAI格式为例，实际需替换为你的API密钥和Base URL） client = openai.OpenAI( api_key="YOUR_API_KEY", base_url="https://api.openai.com/v1" # 或国内模型的对应地址 ) try: response = client.chat.completions.create( model="gpt-3.5-turbo", # 或 "gpt-4", "claude-3-haiku"等 messages=[ {"role": "system", "content": "你是一个专业的金融分析助手。"}, {"role": "user", "content": prompt} ], max_tokens=300, temperature=0.7, ) comment = response.choices[0].message.content.strip() return {"comment": comment} except Exception as e: # 如果大模型调用失败，返回一个兜底的简单总结 industry_count = {} for s in screened_stocks: ind = s.get('industry', '未知') industry_count[ind] = industry_count.get(ind, 0) + 1 top_industry = max(industry_count.items(), key=lambda x: x[1]) if industry_count else ('无', 0) return {"comment": f"模型调用失败，使用备用总结。本次筛选出{len(screened_stocks)}只股票，主要分布在{top_industry[0]}行业（{top_industry[1]}只）。"}

4.3 Agent的完整实现与编排

现在，我们将这些Skill组合起来，形成一个强大的EnhancedScreeningAgent。

# enhanced_screening_agent.py import asyncio import yaml from typing import List, Dict, Any from openclaw import Agent from advanced_data_skill import get_stock_daily from technical_skill import calculate_technical_indicators from llm_report_skill import generate_stock_comment # 假设还有其他Skill，如 get_stock_valuation_batch, screen_by_financials等 class EnhancedScreeningAgent(Agent): name = "enhanced_value_screening_agent" description = "增强版价值筛选Agent，包含多因子筛选、技术指标和大模型点评" def __init__(self, config_path: str = "config/screening_config.yaml"): super().__init__() self.config = self._load_config(config_path) def _load_config(self, path): with open(path, 'r', encoding='utf-8') as f: return yaml.safe_load(f) async def run(self): self.logger.info("开始增强版选股流程...") results = {} # 步骤1: 获取待分析的股票池 (这里简化，从配置读取) watch_list = self.config.get('watch_list', []) if not watch_list: self.logger.error("监控列表为空，请检查配置。") return results # 步骤2: 并发获取批量数据 self.logger.info(f"开始并发获取 {len(watch_list)} 只股票的数据...") data_tasks = [] for code in watch_list[:50]: # 限制数量，防止请求过多 task = get_stock_daily(stock_code=code, start_date="20240101") data_tasks.append(task) daily_data_results = await asyncio.gather(*data_tasks, return_exceptions=True) # 步骤3: 数据处理与因子计算 screened_candidates = [] for i, result in enumerate(daily_data_results): if isinstance(result, Exception): self.logger.warning(f"获取股票{watch_list[i]}数据时出错: {result}") continue daily_data = result.get('daily_data') if not daily_data: continue # 计算技术指标 indicators_result = calculate_technical_indicators( daily_data=daily_data, indicators=['ma20', 'ma60', 'rsi14'] ) tech_data = indicators_result.get('indicators_data', {}) # 这里应插入获取财务数据的Skill调用（示例省略） # financial_result = await get_stock_financials(watch_list[i]) # financial_data = financial_result.get('financials', {}) # 步骤4: 应用筛选规则 (示例：收盘价在20日均线以上，且RSI<70) close_price = daily_data[-1]['close'] if daily_data else None ma20 = tech_data.get('ma20') rsi14 = tech_data.get('rsi14') screening_rules = self.config.get('screening_rules', {}) price_above_ma = screening_rules.get('price_above_ma', True) rsi_threshold = screening_rules.get('rsi_threshold', 70) pass_screen = True if price_above_ma and (ma20 is None or close_price is None or close_price <= ma20): pass_screen = False if rsi14 is not None and rsi14 >= rsi_threshold: pass_screen = False if pass_screen: screened_candidates.append({ 'code': watch_list[i], 'close': close_price, 'ma20': ma20, 'rsi14': rsi14, # ... 其他财务数据 }) # 步骤5: 排序 (例如按RSI升序，越低越可能超卖) screened_candidates.sort(key=lambda x: x.get('rsi14', 100)) final_list = screened_candidates[:self.config.get('output_limit', 10)] # 步骤6: 调用大模型生成评论 strategy_desc = self.config.get('strategy_description', '价格在20日均线上且RSI低于70的技术性筛选') comment_result = await generate_stock_comment(final_list, strategy_desc) comment = comment_result.get('comment', '') # 步骤7: 整合结果并输出 results = { 'screening_date': asyncio.get_event_loop().time(), 'strategy': strategy_desc, 'qualified_stocks': final_list, 'model_comment': comment } # 可以在这里调用一个输出Skill，将results保存为JSON或CSV，或发送通知 # await output_results_skill(results) self.logger.info(f"流程结束。筛选出 {len(final_list)} 只股票。") self.logger.info(f"模型点评：{comment[:100]}...") # 打印前100字符 return results # 配置文件 screening_config.yaml 示例 """ watch_list: - "000001" # 平安银行 - "000002" # 万科A - "600519" # 贵州茅台 - "300750" # 宁德时代 screening_rules: price_above_ma: true rsi_threshold: 70 output_limit: 15 strategy_description: “价格位于20日均线以上且RSI低于70的短线技术筛选策略” """

这个Agent展示了更复杂的逻辑：配置化、并发处理、多因子判断、集成大模型。你可以通过修改YAML配置文件来调整策略，而无需改动代码。

5. 部署、调优与避坑指南

让系统在服务器上稳定运行，并处理好各种边缘情况，才是项目从“玩具”到“工具”的关键。

5.1 使用Docker容器化部署

Docker能确保环境一致性，方便迁移和扩展。创建一个Dockerfile：

# 使用官方Python镜像 FROM python:3.10-slim # 设置工作目录 WORKDIR /app # 复制依赖文件并安装 COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple # 复制应用代码 COPY . . # 声明环境变量（如API密钥） # ENV OPENAI_API_KEY="your_key_here" # 运行Agent CMD ["python", "your_main_agent_script.py"]

对应的requirements.txt文件：

openclaw @ git+https://github.com/openclaw-ai/openclaw.git akshare pandas numpy pyyaml openai # 或 zhipuai, dashscope等 aiohttp

然后构建并运行镜像：

docker build -t openclaw-stock-screener . docker run --rm openclaw-stock-screener

5.2 性能优化与稳定性提升

1. 异步并发：如上面Agent所示，使用asyncio.gather并发执行多个数据获取任务，这是提升速度最有效的方法。
2. 缓存策略：对变化不频繁的数据（如股票列表、公司基本信息）实施缓存。可以使用内存缓存（如functools.lru_cache）、文件缓存或Redis。
3. 速率限制与重试：财经数据API通常有调用频率限制。必须实现速率控制（例如使用asyncio.Semaphore）和指数退避的重试机制。
4. 错误隔离：一个Skill的失败不应导致整个Agent崩溃。使用try...except妥善处理异常，并记录日志。
5. 日志记录：使用Python的logging模块，为不同组件设置不同日志级别（INFO, DEBUG, ERROR），并输出到文件，方便后期排查。

5.3 常见问题与排查实录

在实际操作中，你几乎一定会遇到以下问题：

• 问题1：akshare接口返回空数据或报错。

  • 原因：akshare的接口高度依赖第三方网站结构，网站改版会导致接口失效。
  • 排查：首先检查你的akshare是否为最新版本（pip install -U akshare）。其次，直接运行出错的接口函数，看错误信息。最后，查阅akshare的GitHub Issues，看是否有相同问题及解决方案。
  • 解决：临时方案是寻找替代接口（akshare内同功能的其他函数）。长期方案是考虑使用更稳定的付费数据源，或者自己维护爬虫。

• 问题2：大模型API调用超时或返回非预期内容。

  • 原因：网络不稳定、API密钥错误、额度不足、或Prompt设计不佳。
  • 排查：检查网络连通性；确认API密钥和Base URL正确；在控制台查看调用消耗的token和费用；检查返回的完整响应内容。
  • 解决：增加超时设置和重试；使用更明确的System Prompt和User Prompt；对输出内容做后处理校验（如检查是否包含关键词）。

• 问题3：Agent运行一段时间后内存占用越来越高。

  • 原因：可能是缓存未清理，或者在循环中积累了大量的中间数据。
  • 排查：使用memory_profiler等工具定位内存增长点。
  • 解决：对于缓存，设置合理的TTL或大小限制。在循环中及时删除不再需要的大对象。考虑将中间结果持久化到磁盘或数据库，而非一直留在内存。

• 问题4：筛选出的股票数量为0或过多，策略不理想。

  • 原因：这是策略逻辑问题，不是技术问题。筛选条件太严或太松。
  • 排查：回测！将你的策略逻辑应用到历史数据上，看看在历史上的表现如何。这是量化交易的核心步骤。
  • 解决：调整筛选参数（如PE阈值、技术指标参数）。引入更多因子进行综合评判。牢记：这个系统是辅助工具，它的输出必须经过你的人工判断，绝不能直接用于交易决策。

终极心得：构建这样一个系统，技术实现只占一半，另一半是金融逻辑和数据质量。花在理解业务逻辑、清洗验证数据、设计回测框架上的时间，可能远多于写代码的时间。OpenClaw这类框架的价值，在于它让“想法快速变成可执行的自动化流程”这件事变得异常简单。你可以用极低的成本，将脑海中的一个选股思路，在几个小时内变成一个能自动运行、产出报告的原型。这极大地加速了策略的迭代和验证过程。但请永远对市场的复杂性和数据的噪音保持敬畏，你的系统只是一个“增强版的望远镜”，而不是“预测未来的水晶球”。