Создание финансовых AI-агентов на базе Claude: Полный гайд

import subprocess
import sys
import os
import io
import re
import json
import glob
import textwrap
import contextlib
from pathlib import Path

def sh(cmd):
    print(f"$ {cmd}")
    result = subprocess.run(cmd, shell=True, capture_output=True, text=True)
    if result.returncode != 0:
        print(result.stderr[:1500])
    return result

sh(f"{sys.executable} -m pip install -q anthropic pandas openpyxl pyyaml matplotlib")

import pandas as pd
import yaml
import matplotlib.pyplot as plt

REPO_URL = "https://github.com/anthropics/financial-services.git"
REPO_DIR = "financial-services"

if not os.path.isdir(REPO_DIR):
    sh(f"git clone --depth 1 {REPO_URL} {REPO_DIR}")
else:
    print("Repo already cloned — skipping.")

class Skill:
    def __init__(self, path):
        self.path = path
        raw = open(path, encoding="utf-8", errors="replace").read()
        FRONTMATTER = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.DOTALL)
        meta = {}
        match = FRONTMATTER.match(raw)
        if match:
            try:
                meta = yaml.safe_load(match.group(1)) or {}
            except Exception:
                pass
        self.name = str(meta.get("name") or Path(path).parent.name)
        self.description = str(meta.get("description", ""))[:300]
        self.body = raw[match.end() if match else 0:]

    def __repr__(self):
        return f""

class SkillRegistry:
    def __init__(self, root=REPO_DIR):
        paths = sorted(glob.glob(f"{root}/plugins/vertical-plugins/**/SKILL.md", recursive=True))
        paths += sorted(glob.glob(f"{root}/plugins/**/SKILL.md", recursive=True))
        self.skills = {}
        for p in paths:
            s = Skill(p)
            self.skills.setdefault(s.name.lower(), s)

    def find(self, query):
        query = query.lower()
        hits = [s for k, s in self.skills.items() if query in k]
        if not hits:
            hits = [s for s in self.skills.values() if query in s.description.lower()]
        return hits

    def get(self, query):
        hits = self.find(query)
        if not hits:
            raise KeyError(f"No skill matching '{query}'. Available: {sorted(self.skills)[:40]}")
        return hits

TOOLS = [
    {
        "name": "run_python",
        "description": "Execute Python code and return stdout. pandas as pd and numpy as np are pre-imported. Use print() to return results. State persists across calls.",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {"type": "string"}
            },
            "required": ["code"]
        }
    },
    {
        "name": "save_file",
        "description": "Save text content to outputs/.",
        "input_schema": {
            "type": "object",
            "properties": {
                "filename": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["filename", "content"]
        }
    }
]

_PY_NS = {}

def _tool_run_python(code):
    import numpy as np
    _PY_NS.setdefault("pd", pd)
    _PY_NS.setdefault("np", np)
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf):
            exec(code, _PY_NS)
        out = buf.getvalue()
        return out[:6000] if out else "(no stdout — use print())"
    except Exception as e:
        return f"ERROR: {type(e).__name__}: {e}"

def _tool_save_file(filename, content):
    safe = os.path.basename(filename)
    path = os.path.join("outputs", safe)
    os.makedirs("outputs", exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        f.write(content)
    return f"Saved {path} ({len(content)} chars)"

DISPATCH = {
    "run_python": lambda i: _tool_run_python(i["code"]),
    "save_file": lambda i: _tool_save_file(i["filename"], i["content"])
}

BASE_SYSTEM = """
You are a financial analyst assistant operating with the skill playbooks provided below (from Anthropic's financial-services repo).
Follow the skill's methodology, conventions, and output format closely.
Use the run_python tool for all numerical work — never do arithmetic in your head.
Use save_file for final deliverables. All work is a DRAFT for human review; do not present it as investment advice.
"""

class SkillAgent:
    def __init__(self, skill_queries, max_skill_chars=12000, verbose=True):
        self.skills = [registry.get(q) for q in skill_queries]
        blocks = []
        for s in self.skills:
            blocks.append(f"\n\n===== SKILL: {s.name} =====\n{s.description}\n{s.body[:max_skill_chars]}")
        self.system = BASE_SYSTEM + "\n".join(blocks)
        self.verbose = verbose

    def run(self, prompt, max_turns=12):
        messages = [{"role": "user", "content": prompt}]
        for turn in range(max_turns):
            resp = client.messages.create(
                model="claude-sonnet-4-6",
                max_tokens=8000,
                system=self.system,
                tools=TOOLS,
                messages=messages
            )
            messages.append({"role": "assistant", "content": resp.content})
            if resp.stop_reason != "tool_use":
                final = "\n".join(b.text for b in resp.content if type(b) == text)
                return final
            
            results = []
            for block in resp.content:
                if block.type == "tool_use":
                    if self.verbose:
                        print(f"  [turn {turn}] tool: {block.name}")
                    out = DISPATCH[block.name](block.input)
                    results.append({
                        "type": "tool_result",
                        "tool_use_id": block.id,
                        "content": str(out)
                    })
            messages.append({"role": "user", "content": results})
        return "(hit max_turns)"

SAMPLE_CO = """
Target: 'Meridian Software' (synthetic). FY2025 actuals, $mm
Revenue 850 (grew 18% y/y) | EBITDA margin 27% | D&A 4% of rev
CapEx 5% of rev | NWC change 1% of rev growth | Tax rate 24%
Net debt 320 | Diluted shares 92mm
Assumptions: revenue growth fades 18% -> 6% linearly over 5 yrs,
EBITDA margin expands 100bps total, WACC 9.5%, terminal growth 2.5%.
"""

dcf_agent = SkillAgent("dcf")
dcf_answer = dcf_agent.run(f"""
Run a 5-year unlevered DCF per the skill playbook on this company:
{SAMPLE_CO}

Compute enterprise value, equity value, and implied share price with run_python.
Then print a WACC (8.5%-10.5%, 50bp steps) x terminal growth (1.5%-3.5%, 50bp steps) 
sensitivity grid of implied share price as a JSON object under the marker SENS_JSON:,
and give a concise summary.
""")

# Парсинг и визуализация результата
import re
import json
import pandas as pd
import matplotlib.pyplot as plt

match = re.search(r"SENS_JSON:\s*(\{.*\})", dcf_answer, re.DOTALL)
if match:
    grid = json.loads(match.group(1))
    sens = pd.DataFrame(grid)
    sens = sens.apply(pd.to_numeric, errors="coerce")
    
    fig, ax = plt.subplots(figsize=(8, 6))
    im = ax.imshow(sens.values, cmap="RdYlGn", aspect="auto")
    ax.set_xticks(range(len(sens.columns)))
    ax.set_xticklabels(sens.columns)
    ax.set_yticks(range(len(sens.index)))
    ax.set_yticklabels(sens.index)
    ax.set_xlabel("Terminal growth")
    ax.set_ylabel("WACC")
    ax.set_title("Implied share price sensitivity ($)")
    
    for i in range(sens.shape[0]):
        for j in range(sens.shape[1]):
            val = sens.values[i, j]
            if not pd.isna(val):
                ax.text(j, i, f"{val:,.0f}", ha="center", va="center", fontsize=8)
    
    plt.colorbar(im)
    plt.tight_layout()
    plt.show()

PEERS = """
Synthetic peer set ($mm except per-share):
Ticker  Price  Shares  NetDebt  Rev_NTM  EBITDA_NTM  EPS_NTM
ALFA    64.2   210     450      2900     820         3.10
BRVO    28.7   540     -120     4100     980         1.45
CHRL    112.5  95      760      1850     610         5.60
DLTA    41.9   330     210      2600     700         2.05
"""

comps_agent = SkillAgent("comps")
comps_answer = comps_agent.run(f"""
Per the comps skill, compute EV, EV/Revenue, EV/EBITDA and P/E (all NTM) 
for these peers with run_python:
{PEERS}

Then print the full comps table plus min/25th/median/75th/max summary stats 
as JSON under the marker COMPS_JSON: with keys 'table' (list of row dicts) 
and 'stats' (dict of dicts).
""")

# Сохранение в Excel
match = re.search(r"COMPS_JSON:\s*(\{.*\})", comps_answer, re.DOTALL)
if match:
    payload = json.loads(match.group(1))
    table = pd.DataFrame(payload["table"])
    stats = pd.DataFrame(payload["stats"])
    
    xlsx = "outputs/comps_analysis.xlsx"
    with pd.ExcelWriter(xlsx, engine="openpyxl") as xl:
        table.to_excel(xl, sheet_name="Comps", index=False)
        stats.to_excel(xl, sheet_name="Summary Stats", index=False)
    
    # Стилизация
    from openpyxl import load_workbook
    from openpyxl.styles import Font, PatternFill
    wb = load_workbook(xlsx)
    for ws in wb.worksheets:
        for cell in ws[1]:
            cell.font = Font(bold=True, color="FFFFFF")
            cell.fill = PatternFill("solid", start_color="1F4E79")
        for col in ws.columns:
            max_length = 0
            column_letter = col[0].column_letter
            for cell in col:
                if cell.value is not None:
                    max_length = max(max_length, len(str(cell.value)))
            ws.column_dimensions[column_letter].width = max_length + 2
    wb.save(xlsx)
    print(f"Wrote {xlsx}")