evotraders/backend/agents/toolkit_factory.py

# -*- coding: utf-8 -*-
"""Toolkit factory following AgentScope's skill + tool group practices."""

from typing import Any, Dict, Iterable

from .agent_workspace import load_agent_workspace_config
from backend.config.bootstrap_config import get_bootstrap_config_for_run
import yaml

from .skills_manager import SkillsManager


def load_agent_profiles() -> Dict[str, Dict[str, Any]]:
    config_path = SkillsManager().project_root / "backend" / "config" / "agent_profiles.yaml"
    with open(config_path, "r", encoding="utf-8") as file:
        return yaml.safe_load(file) or {}


def _register_analysis_tool_groups(toolkit: Any) -> None:
    from backend.tools.analysis_tools import TOOL_REGISTRY

    tool_groups = {
        "fundamentals": {
            "description": "Financial health, profitability, growth, and efficiency analysis tools.",
            "active": False,
            "notes": (
                "Use these tools to validate business quality, financial resilience, "
                "and earnings durability before making directional conclusions."
            ),
            "tools": [
                "analyze_profitability",
                "analyze_growth",
                "analyze_financial_health",
                "analyze_efficiency_ratios",
                "analyze_valuation_ratios",
                "get_financial_metrics_tool",
            ],
        },
        "technical": {
            "description": "Trend, momentum, mean reversion, and volatility analysis tools.",
            "active": False,
            "notes": (
                "Use these tools to assess timing, price structure, and risk-reward in "
                "the current market regime."
            ),
            "tools": [
                "analyze_trend_following",
                "analyze_momentum",
                "analyze_mean_reversion",
                "analyze_volatility",
            ],
        },
        "sentiment": {
            "description": "News sentiment and insider activity analysis tools.",
            "active": False,
            "notes": (
                "Use these tools to capture short-horizon catalysts, sentiment shifts, "
                "and behavioral signals around each ticker."
            ),
            "tools": [
                "analyze_news_sentiment",
                "analyze_insider_trading",
            ],
        },
        "valuation": {
            "description": "Intrinsic value and relative valuation analysis tools.",
            "active": False,
            "notes": (
                "Use these tools when the task requires fair value estimation, margin of "
                "safety analysis, or valuation scenario comparison."
            ),
            "tools": [
                "dcf_valuation_analysis",
                "owner_earnings_valuation_analysis",
                "ev_ebitda_valuation_analysis",
                "residual_income_valuation_analysis",
            ],
        },
    }

    for group_name, group_config in tool_groups.items():
        toolkit.create_tool_group(
            group_name=group_name,
            description=group_config["description"],
            active=group_config["active"],
            notes=group_config["notes"],
        )
        for tool_name in group_config["tools"]:
            tool_func = TOOL_REGISTRY.get(tool_name)
            if tool_func:
                toolkit.register_tool_function(
                    tool_func,
                    group_name=group_name,
                )


def _register_portfolio_tool_groups(toolkit: Any, pm_agent: Any) -> None:
    toolkit.create_tool_group(
        group_name="portfolio_ops",
        description="Portfolio decision recording tools.",
        active=False,
        notes=(
            "Use portfolio tools only after synthesizing analyst and risk inputs. "
            "Record one explicit decision per ticker."
        ),
    )
    toolkit.register_tool_function(
        pm_agent._make_decision,
        group_name="portfolio_ops",
    )


def _register_risk_tool_groups(toolkit: Any) -> None:
    from backend.tools.risk_tools import (
        assess_margin_and_liquidity,
        assess_position_concentration,
        assess_volatility_exposure,
    )

    toolkit.create_tool_group(
        group_name="risk_ops",
        description="Risk diagnostics for concentration, leverage, and volatility.",
        active=False,
        notes=(
            "Use risk tools to quantify concentration, margin pressure, and volatility "
            "before writing the final risk memo."
        ),
    )
    toolkit.register_tool_function(
        assess_position_concentration,
        group_name="risk_ops",
    )
    toolkit.register_tool_function(
        assess_margin_and_liquidity,
        group_name="risk_ops",
    )
    toolkit.register_tool_function(
        assess_volatility_exposure,
        group_name="risk_ops",
    )


def create_agent_toolkit(
    agent_id: str,
    config_name: str,
    owner: Any = None,
    active_skill_dirs: Iterable[str] | None = None,
) -> Any:
    """Create a Toolkit with agent skills and grouped tools."""
    from agentscope.tool import Toolkit

    profiles = load_agent_profiles()
    profile = profiles.get(agent_id, {})
    skills_manager = SkillsManager()
    agent_config = load_agent_workspace_config(
        skills_manager.get_agent_asset_dir(config_name, agent_id) / "agent.yaml",
    )
    bootstrap_config = get_bootstrap_config_for_run(
        skills_manager.project_root,
        config_name,
    )
    override = bootstrap_config.agent_override(agent_id)
    active_groups = override.get(
        "active_tool_groups",
        agent_config.active_tool_groups
        or profile.get("active_tool_groups", []),
    )
    disabled_groups = set(agent_config.disabled_tool_groups)
    if disabled_groups:
        active_groups = [
            group_name
            for group_name in active_groups
            if group_name not in disabled_groups
        ]

    toolkit = Toolkit(
        agent_skill_instruction=(
            "<system-info>You have access to project skills. Each skill lives in a "
            "directory and is described by SKILL.md. Follow the skill instructions "
            "when they are relevant to the current task.</system-info>"
        ),
        agent_skill_template="- {name} (dir: {dir}): {description}",
    )

    if agent_id.endswith("_analyst"):
        _register_analysis_tool_groups(toolkit)
    elif agent_id == "portfolio_manager" and owner is not None:
        _register_portfolio_tool_groups(toolkit, owner)
    elif agent_id == "risk_manager":
        _register_risk_tool_groups(toolkit)

    if active_skill_dirs is None:
        skill_names = skills_manager.resolve_agent_skill_names(
            config_name=config_name,
            agent_id=agent_id,
            default_skills=profile.get("skills", []),
        )
        active_skill_dirs = [
            skills_manager.get_agent_active_root(config_name, agent_id) / skill_name
            for skill_name in skill_names
        ]

    for skill_dir in active_skill_dirs:
        toolkit.register_agent_skill(str(skill_dir))

    if active_groups:
        toolkit.update_tool_groups(group_names=active_groups, active=True)

    return toolkit