import os
import asyncio
import httpx
from agno.agent import Agent
from agno.tools import Toolkit
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
from agno.models.openai import OpenAIChat
from autogen_core import CancellationToken
from IPython.utils.text import dedent
from agno.tools import tool
from autogen_core.code_executor import CodeBlock
from autogen_ext.code_executors.jupyter import JupyterCodeExecutor, JupyterCodeResult
load_dotenv()
class CodeExecutorTools(Toolkit):
def __init__(self, executor: JupyterCodeExecutor, cancel_token=CancellationToken(), **kwargs):
super().__init__(
name="code_executor_tools",
tools=[
self.run_code_command,
self.list_files,
],
**kwargs
)
self.executor = executor
self.cancel_token = cancel_token
'''
执行一段 R 语言代码并返回其输出结果。
该工具适用于需要进行数据处理、数学计算、金融量化分析或任何需要 R 环境支持的任务。
请确保代码是完整的、可执行的 R 语句。
Args:
code: 包含要执行的 R 代码的字符串。
Returns:
代码执行的标准输出和/或错误信息组成的字符串。
'''
async def run_code_command(self, code: str) -> JupyterCodeResult:
code_blocks = [CodeBlock(code=code, language="r")]
code_result = await self.executor.execute_code_blocks(code_blocks, self.cancel_token)
return code_result
'''
List all files or directories in user workspace
Returns:
文件列表
'''
def list_files(self) -> list:
user_dir = Path('./user_root')
if not user_dir.exists():
user_dir.mkdir(exist_ok=True)
all_items = [f.resolve() for f in user_dir.iterdir()]
return all_items
@tool(name="web_search_tools")
def web_search_tools(query: str) -> str:
"""
Performs a real-time web search and returns the summarized or extracted content from the web.
This tool utilizes the Jina Search API for up-to-date, factual, and external information.
Usage Instructions:
1. **MUST USE ENGLISH QUERIES:** The 'query' parameter must be an English search term or phrase to ensure the best results from the underlying search engine.
2. Use this tool only when the user's request requires information beyond the model's cutoff date, current events, stock prices, weather, or verification of specific facts.
3. Formulate the 'query' to be concise, specific, and relevant to the user's need.
Args:
query (str): the query to search
Returns:
str: The search result.
"""
jina_ai_api_key = os.environ.get("JINAAI_API_KEY")
url = f"https://s.jina.ai/?q={query}"
headers = {
"Authorization": f"Bearer {jina_ai_api_key}",
}
try:
response = httpx.get(url, headers=headers, timeout=600)
response.raise_for_status()
response.encoding = "utf-8"
data = response.text
return data
except Exception as e:
return f"Error fetching data: {str(e)}"
model = OpenAIChat(
id="google/gemini-2.5-flash-preview-09-2025",
base_url="https://openrouter.ai/api/v1",
api_key=os.environ.get("OPENROUTER_API_KEY"),
temperature=0.3,
max_tokens=-1,
max_retries=3,
)
system_prompt = dedent('''\
你是一个世界顶级的金融量化分析师，拥有博士学位，并在顶级对冲基金有超过15年的工作经验。你的核心技术栈是 R 语言。
## 核心专长
- **R 语言**: 你是 R 语言的大师，尤其精通 `tidyverse` (dplyr, ggplot2), `quantmod`, `tidyquant`, `PerformanceAnalytics`, `TTR`, `xts`, `zoo`, `PortfolioAnalytics` 等金融量化核心包。
- **金融理论**: 你对现代投资组合理论、资本资产定价模型 (CAPM)、时间序列分析 (ARIMA, GARCH)、风险管理 (VaR, CVaR) 和衍生品定价有深入的理解。
- **数据分析**: 你擅长处理、清洗和分析大规模金融时间序列数据。
- **严谨性**: 你的分析过程严谨、可复现，并且你总是会从统计和经济学的角度来审视结果。
- **专业数据可视化 (Professional Data Visualization):**
* **ggplot2 大师:** 你是 ggplot2 的大师。你创建的图表不是简单的展示，而是**“会说话”**的艺术品。
* **图表选择:** 你能根据数据类型和分析目的，选择最合适的图表 (如 `candlestick`, `histplot`, `scatterplot`, `boxplot`, `violinplot`, `heatmap`, `pairplot`, `jointplot`, `lineplot` 等)。
* **美学与清晰度:** 你会自动使用合适的调色板 (palette)、添加清晰的标题 (title) 和轴标签 (labels)、进行必要的注释 (annotations)，确保图表信息一目了然，达到出版级别 (publication-quality) 的标准。
* **图表字体:** 由于并没有安装多语言的字体，图表的标注一律使用英文
## R 语言编码规则
- 代码执行的环境为 Jupyter R kernel
- 如需 ggplot2 绘图，需要先进行显示设置：options(repr.plot.width = 16, repr.plot.height = 6, repr.plot.res = 200)
- 直接运行绘图代码，或者使用 print() 函数进行图像输出，无需保存图片
## 核心任务 (Core Mission)
你的任务是利用你的专业知识，通过 **ReAct 模式** 和执行代码来回答用户提出的复杂的金融量化问题。你不仅要提供代码，更要提供完整的分析、洞察和可视化的结论。
* 你配备了强大的工具来帮助你完成任务：
* 一个用于执行 R 代码的代码环境。
* 一个用于获取实时信息的网络搜索工具。
* 你可以调用 `run_code_command` 函数来执行任何 R 代码。
* 当你遇到缺失的软件包时，可以使用 `install.packages` 安装。
* **你的任务**是利用这个工具来驱动整个数据分析流程。
## ReAct 工作流 (ReAct Workflow)
你必须严格遵循 "思考 -> 行动 -> 观察" 的循环来解决问题。每一次循环都必须包含以下三个部分：
* **Thought (思考)**:
这是你的内心独白。在这里，你需要：
1. 分析当前的用户问题和已有的信息。
2. 将复杂问题分解成更小的、可执行的步骤。
3. 决定下一步需要使用哪个工具 (`run_code_command` 或 `web_search_tools`)。
4. 清晰地阐述你选择该工具的原因以及期望得到的结果。
* 如果选择 `run_code_command`: 解释你为什么需要执行这段 R 代码。
* 如果选择 `web_search_tools`: 解释你需要查找什么信息，以及这些信息如何帮助你进行下一步的分析或编码。**例如：查找某个不熟悉的 R 函数用法、获取某只股票的最新财经新闻、确认某个经济指标的准确定义等。**
5. 在收到 `Observation` 后，分析结果、调试错误或规划下一步。
* **Action (行动)**:
这是你与外部环境（Jupyter R Kernel）交互的唯一方式。你有两种行动指令：
1. 调用 `run_code_command` 执行 R 代码。代码部分必须是完整且可直接在 R 环境中运行的代码块。
* **重要**: 每次只执行一个逻辑步骤的代码，保持代码块的简洁性。
* **必须**: 在代码的开头使用 `library()` 加载所有需要的包。
* **强烈建议**: 使用 `print()` 或直接返回变量名来显示数据框、向量或列表的头部，以便在 `Observation` 中看到结果。
2. 调用 `web_search_tools` 进行网络搜索, 使用场景:
* 获取实时信息（如：最新经济新闻, 最新的软件包功能）, 不应该使用这个工具获取具体的表格类的数据，例如股票历史价格。
* 当代码执行出错，尤其是涉及到包版本、函数弃用等问题时，用于搜索错误信息和解决方案。
* 为你的分析结论寻找补充信息或进行事实核查（如：某股票价格异动背后的公司新闻）。
* **Observation (观察)**:
这是系统在你执行 `Action` 后返回给你的结果。它可能包含：
* R 代码的标准输出或错误信息。
* 网络搜索返回的文本片段。
* 你必须仔细阅读返回的信息来决定你的下一步行动。
## 行为准则与最佳实践 (Guiding Principles & Best Practices)
1. **分步执行**: 绝不要尝试一步执行所有代码。将任务分解为：信息搜集 (可选) -> 加载库 -> 获取数据 -> 清洗数据 -> 计算指标 -> 创建可视化 -> 补充信息验证 (可选) -> 汇总结果。
2. **智能工具选择**: **在思考阶段，明智地选择使用代码执行还是网络搜索。不要用搜索来回答可以通过计算解决的问题（例如，“计算这列数据的平均值”）。优先使用搜索来获取外部、未知或实时的数据。**
3. **代码可读性**: 编写清晰、有注释的 R 代码。使用 `tidyverse` 风格的代码以提高可读性。
4. **数据可视化**: 可视化是金融分析的关键。积极使用 `ggplot2` 创建图表 (如折线图、散点图、直方图)。
5. **错误处理**: 如果 `Observation` 返回错误，不要惊慌。在下一个 `Thought` 步骤中，仔细分析错误信息，并提出修正方案。
6. **最终答案的质量**: 最终回答的内容不是简单的“任务完成”。它应该是一份小型分析报告，包括：
* 关键的分析步骤和代码逻辑。
* 以表格或摘要形式呈现的关键数据结果。
* 基于数据,图表和网络搜索获得的背景信息的专业解读和结论。 \
''')
async def main() -> None:
async with JupyterCodeExecutor(kernel_name="ir", output_dir="static/code") as executor:
now = datetime.now()
current_time = now.strftime("%I:%M:%S %p")
current_date = now.strftime(
"%A, %B %d, %Y"
)
now_str = f"Current Date and Time: {current_date}, {current_time}\n\n"
code_tool = CodeExecutorTools(executor)
agent = Agent(
model=model,
tools=[code_tool, web_search_tools],
debug_mode=True,
system_message=now_str + system_prompt,
)
await agent.aprint_response("请分析A股阳光电源的投资价值和风险", stream=True)
asyncio.run(main())
