aksharetools：大模型智能体框架agno可直接获取A股金融数据

原创内容第889篇，专注智能量化投资、个人成长与财富自由。

今天说说金融智能体开发。

智能体开发需要一个多agent框架。这样的框架，现在太多了，langchain, langgraph，autogen，crewai等等，还有各种低代码平台。

我找到一个轻量级的多模态的智能体框架平台：Agno。

使用akshare封装成函数集：AShareTools，这样对大模型直接提问：

from agno.agent import Agentfrom agno.tools.yfinance import YFinanceToolsfrom akshare_tools import AShareToolsfrom utils import get_modelagent = Agent(model=get_model(name='qwen'),                tools=[AShareTools(enable_all=True)],                 instructions=[                         "Use tables to display data.",                         "Only include the table in your response. No other text.",                     ],              debug_mode=False,              )# Getting chunks in a variablerun_response = agent.run("获取sz000001,sh600519的最新股价", stream=True)for chunk in run_response:    for char in chunk.content:        print(char, end='', flush=True)  # 关键参数：end=''取消换行，flush=True立即刷新        #time.sleep(0.05)  # 调整这个值控制打字速度（单位：秒）

import jsonfrom datetime import datetime, timedeltaimport pandas as pdimport akshare as akfrom agno.tools import Toolkitfrom agno.utils.log import log_debugclass AShareTools(Toolkit):    """    AShareTools is a toolkit for getting financial data from A-share markets using akshare.    Note: Symbols should include market prefix (e.g., sh600000 for Shanghai, sz000001 for Shenzhen).    Args:        stock_price (bool): Whether to get the current stock price.        company_info (bool): Whether to get company information.        stock_fundamentals (bool): Whether to get stock fundamentals.        income_statements (bool): Whether to get income statements.        key_financial_ratios (bool): Whether to get key financial ratios.        analyst_recommendations (bool): Whether to get analyst recommendations.        company_news (bool): Whether to get company news.        technical_indicators (bool): Whether to get technical indicators.        historical_prices (bool): Whether to get historical prices.        enable_all (bool): Whether to enable all tools.    """    def __init__(            self,            stock_price: bool = True,            company_info: bool = False,            stock_fundamentals: bool = False,            income_statements: bool = False,            key_financial_ratios: bool = False,            analyst_recommendations: bool = False,            company_news: bool = False,            technical_indicators: bool = False,            historical_prices: bool = False,            enable_all: bool = False,            **kwargs,    ):        super().__init__(name="ashare_tools", **kwargs)        if stock_price or enable_all:            self.register(self.get_current_stock_price)        if company_info or enable_all:            self.register(self.get_company_info)        if stock_fundamentals or enable_all:            self.register(self.get_stock_fundamentals)        if income_statements or enable_all:            self.register(self.get_income_statements)        if key_financial_ratios or enable_all:            self.register(self.get_key_financial_ratios)        if analyst_recommendations or enable_all:            self.register(self.get_analyst_recommendations)        if company_news or enable_all:            self.register(self.get_company_news)        if technical_indicators or enable_all:            self.register(self.get_technical_indicators)        if historical_prices or enable_all:            self.register(self.get_historical_stock_prices)    def get_current_stock_price(self, symbol: str) -> str:        code = symbol[2:]        """        Get the current stock price for a given A-share symbol.        Args:            symbol (str): The stock symbol with market prefix (e.g., sh600000).        Returns:            str: Current price formatted to 4 decimal places or error message.        """        try:            log_debug(f"Fetching current price for {symbol}")            # Get all A-share spot data and filter by symbol            df = ak.stock_zh_a_spot_em()            #print(df)            sub = df[df['代码'] == code]['最新价']            #print(sub)            current_price = sub.values[0]            #print(current_price)            return f"{current_price:.4f}" if current_price else f"Could not fetch price for {symbol}"        except Exception as e:            return f"Error fetching current price for {symbol}: {e}"    def get_company_info(self, symbol: str) -> str:        """        Get company information for a given A-share symbol.        Args:            symbol (str): The stock symbol with market prefix.        Returns:            str: JSON containing company profile.        """        try:            code = symbol[2:]  # Remove market prefix            info_df = ak.stock_individual_info_em(symbol=code)            info = dict(zip(info_df['item'], info_df['value']))            # Get real-time price separately            price_df = ak.stock_zh_a_spot_em()            price_data = price_df[price_df['代码'] == symbol].iloc[0]            company_info = {                "Name": info.get('公司名称', 'N/A'),                "Symbol": symbol,                "Current Price": f"{price_data['最新价']} CNY",                "Sector": info.get('所属行业', 'N/A'),                "Industry": info.get('行业分类', 'N/A'),                "Market Cap": f"{price_data['总市值']} CNY",                "PE Ratio": info.get('市盈率', 'N/A'),                "EPS": info.get('每股收益', 'N/A'),                "Website": info.get('公司主页', 'N/A'),                "Business Scope": info.get('经营范围', 'N/A')            }            return json.dumps(company_info, indent=2, ensure_ascii=False)        except Exception as e:            return f"Error fetching company info for {symbol}: {e}"    def get_historical_stock_prices(self, symbol: str, period: str = "1mo", interval: str = "1d") -> str:        """        Get historical prices for a given A-share symbol.        Args:            symbol (str): Stock symbol with market prefix.            period (str): Historical period (1d, 1mo, etc.).            interval (str): Data interval (not fully implemented for akshare).        Returns:            str: JSON formatted historical data.        """        try:            log_debug(f"Fetching historical prices for {symbol}")            code = symbol[2:]            # Convert period to start/end dates            end_date = datetime.now().strftime("%Y%m%d")            period_map = {                "1d": 1, "5d": 5, "1mo": 30,                "3mo": 90, "6mo": 180, "1y": 365            }            start_date = (datetime.now() - timedelta(days=period_map.get(period, 30))).strftime("%Y%m%d")            df = ak.stock_zh_a_daily(symbol=code, start_date=start_date, end_date=end_date, adjust="qfq")            return df[['日期', '开盘', '收盘', '最高', '最低', '成交量']].to_json(orient="records")        except Exception as e:            return f"Error fetching historical prices for {symbol}: {e}"    def get_stock_fundamentals(self, symbol: str) -> str:        """        Get fundamental data for a given A-share symbol.        Args:            symbol (str): Stock symbol with market prefix.        Returns:            str: JSON formatted fundamental data.        """        try:            code = symbol[2:]            indicator_df = ak.stock_a_lg_indicator(stock=code)            latest = indicator_df.iloc[-1].to_dict()            fundamentals = {                "symbol": symbol,                "PE Ratio": latest.get('市盈率'),                "PB Ratio": latest.get('市净率'),                "Dividend Yield": latest.get('股息率'),                "ROE": latest.get('净资产收益率'),                "Total Market Cap (CNY)": latest.get('总市值')            }            return json.dumps(fundamentals, indent=2, ensure_ascii=False)        except Exception as e:            return f"Error getting fundamentals for {symbol}: {e}"    def get_income_statements(self, symbol: str) -> str:        """        Get income statements for a given A-share symbol.        Args:            symbol (str): Stock symbol with market prefix.        Returns:            str: JSON formatted income statements.        """        try:            code = symbol[2:]            df = ak.stock_financial_report_sina(stock=code, symbol="利润表")            return df.to_json(orient="records")        except Exception as e:            return f"Error fetching income statements for {symbol}: {e}"    def get_key_financial_ratios(self, symbol: str) -> str:        """        Get key financial ratios for a given A-share symbol.        Args:            symbol (str): Stock symbol with market prefix.        Returns:            str: JSON formatted financial ratios.        """        try:            code = symbol[2:]            df = ak.stock_financial_analysis_indicator(symbol=code)            return df.to_json(orient="records")        except Exception as e:            return f"Error fetching financial ratios for {symbol}: {e}"    def get_analyst_recommendations(self, symbol: str) -> str:        """        Get analyst recommendations (placeholder - akshare lacks direct equivalent).        Args:            symbol (str): Stock symbol.        Returns:            str: JSON formatted message.        """        return json.dumps({"info": "Analyst recommendations not available via akshare"}, indent=2)    def get_company_news(self, symbol: str, num_stories: int = 3) -> str:        """        Get company news for a given A-share symbol.        Args:            symbol (str): Stock symbol.            num_stories (int): Number of news items to return.        Returns:            str: JSON formatted news items.        """        try:            code = symbol[2:]            news_df = ak.stock_news_em(symbol=code)            return news_df.head(num_stories).to_json(orient="records")        except Exception as e:            return f"Error fetching news for {symbol}: {e}"    def get_technical_indicators(self, symbol: str, period: str = "3mo") -> str:        """        Get technical indicators for a given A-share symbol.        Args:            symbol (str): Stock symbol.            period (str): Historical period.        Returns:            str: JSON formatted technical data.        """        try:            code = symbol[2:]            df = ak.stock_zh_a_daily(symbol=code, period=period, adjust="qfq")            return df[['日期', '收盘', '成交量', 'MA5', 'MA10', 'MA20']].to_json(orient="records")        except Exception as e:            return f"Error fetching technical indicators for {symbol}: {e}"

AGI相关的代码在AGI星球更新：

昨天咱们更新了aitrader_core的1.1版本：年化收益226.9%，最大回撤12.7%，本地运行toml策略，aitrader核心代码发布（python代码+数据下载）

有些同学问，与完全开源的aitrader框架版本的区别？这里做一下说明：框架版本基于wxpython做智能量化平台的gui，回测数据和引擎都是通过api在服务器上完成。

涉及到回测引擎（因子表达式，deap因子挖掘，机器学习建模以及回测引擎等核心代码，每周五在星球更新）

星球部分策略，设定了需要积分才能下载。这个积分是给策略的作者。

如何获得积分？
1、用户加入星球，并绑定会员之后，会获得20积分。每次续费再新增20积分。
3、发布策略，设定“积分查看”，其他同学下载后，积分归作者所有。
3、在论坛发布高质量的贴子即可（可获5，10，15，20等积分）。
4、给同学提供高质量的答案。

比如今天新发布的两个策略：

代码和策略下载：AI量化实验室——2025量化投资的星辰大海

扩展  •  历史文章   

EarnMore（赚得更多）基于RL的投资组合管理框架：一致的股票表示，可定制股票池管理。（附论文+代码）

年化收益200%+的策略集 | 实时板块资金热力图 「aitrader 5.0系统代码发布」

机器学习驱动的策略开发通过流程 | 普通人阶层跃迁的可能路径？

年化30.24%，最大回撤19%，综合动量多因子评分策略再升级（python代码+数据）

三秒钟创建一个年化28%，夏普比1.25的策略（python系统已开放源代码下载）

会员专属策略可以直接下载了，多个十年年化30+%策略集|polars重构因子引擎（代码+数据下载）

6年年化收益46%，最大回撤率为16%的策略（附python代码）