24Esempio 6 · Analisi finanziaria

MODEL=openai/gpt-4.1-mini
OPENAI_API_KEY=sk-la-tua-chiave
SERPER_API_KEY=la-tua-chiave-serper
CREWAI_DISABLE_TELEMETRY=true

anno,ricavi,ebitda,utile_netto,attivo_corrente,passivo_corrente,liquidita,debiti_finanziari,patrimonio_netto
2023,4200,504,168,1650,1100,310,1200,1400
2024,4750,523,143,1720,1350,190,1650,1480
2025,5100,459,92,1800,1640,120,2100,1520

@tool("Calcolatore di indici di bilancio")
def calcola_indici(nome_file: str) -> str:
    """Legge un bilancio in formato CSV (valori in migliaia di euro, una riga per anno) e restituisce
    gli indici principali già calcolati. Usalo sempre invece di fare i conti a mente."""
    nome = Path(nome_file).name  # accetta solo un nome di file della cartella corrente
    if not nome.endswith(".csv") or not Path(nome).exists():
        return f"Errore: il file {nome} non esiste in questa cartella o non è un CSV."

    with open(nome, encoding="utf-8", newline="") as f:
        anni = list(csv.DictReader(f))

    righe = []
    ricavi_prima = None
    for a in anni:
        v = {chiave: float(valore) for chiave, valore in a.items() if chiave != "anno"}
        crescita = percentuale(v["ricavi"] - ricavi_prima, ricavi_prima) if ricavi_prima else "n.d."
        pfn = v["debiti_finanziari"] - v["liquidita"]  # posizione finanziaria netta
        righe.append(
            f"Anno {a['anno']}: ricavi {v['ricavi']:.0f} mila euro (crescita {crescita}); "
            f"margine EBITDA {percentuale(v['ebitda'], v['ricavi'])}; "
            # ... gli altri indici, uguali nella forma
        )
        ricavi_prima = v["ricavi"]
    return "\n".join(righe)

analisi = Task(
    description=(
        "Usa lo strumento Calcolatore di indici di bilancio sul file {file}.\n"
        "Riporta gli indici esattamente come li restituisce lo strumento: non ricalcolarli, "
        "non arrotondarli e non aggiungerne altri. Poi descrivi come sono cambiati negli anni."
    ),
    expected_output=(...),
    agent=analista,
)


def controlla_avviso(risultato: TaskOutput) -> tuple[bool, Any]:
    """Guardrail: la nota deve contenere l'avviso, altrimenti torna all'agente."""
    if AVVISO.lower() not in risultato.raw.lower():
        return (False, f"Manca l'avviso. Inizia la nota con la frase: {AVVISO}.")
    return (True, risultato.raw)

    print("\n=== NOTA (salvata anche in nota_bilancio.md) ===")
    print(risultato.raw)
    # Gli stessi indici calcolati direttamente da Python: confrontali con i numeri della nota.
    print("\n=== INDICI CALCOLATI DA PYTHON, PER IL CONFRONTO ===")
    print(calcola_indici.run(nome_file=FILE_BILANCIO))

analista = Agent(
    role="Analista di notizie finanziarie",
    goal="Raccogliere i fatti recenti e verificabili su {azienda}, ognuno con data e link della fonte",
    backstory=(...),
    tools=[SerperDevTool(search_type="news", n_results=8), ScrapeWebsiteTool()],
    inject_date=True,  # l'agente conosce la data di oggi e può capire cosa è "recente"
    max_iter=10,
    llm=MODELLO,
    verbose=True,
)

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21"]
# ///
"""Python calcola gli indici di un bilancio fittizio, un analista li commenta e un autore scrive la nota."""
import csv
import os
import sys
from pathlib import Path
from typing import Any

from crewai import Agent, Crew, Process, Task, TaskOutput
from crewai.tools import tool

# Il modello si sceglie nel file .env (capitolo 7). Se MODEL manca, si usa quello scritto qui.
MODELLO = os.getenv("MODEL", "openai/gpt-4.1-mini")
FILE_BILANCIO = "bilancio_fittizio_aurora.csv"
AVVISO = "Questa nota non è una consulenza finanziaria"


def percentuale(parte: float, totale: float) -> str:
    return f"{parte / totale * 100:.1f}%" if totale else "n.d."


def rapporto(a: float, b: float) -> str:
    return f"{a / b:.2f}" if b else "n.d."


@tool("Calcolatore di indici di bilancio")
def calcola_indici(nome_file: str) -> str:
    """Legge un bilancio in formato CSV (valori in migliaia di euro, una riga per anno) e restituisce
    gli indici principali già calcolati. Usalo sempre invece di fare i conti a mente."""
    nome = Path(nome_file).name  # accetta solo un nome di file della cartella corrente
    if not nome.endswith(".csv") or not Path(nome).exists():
        return f"Errore: il file {nome} non esiste in questa cartella o non è un CSV."

    with open(nome, encoding="utf-8", newline="") as f:
        anni = list(csv.DictReader(f))

    righe = []
    ricavi_prima = None
    for a in anni:
        v = {chiave: float(valore) for chiave, valore in a.items() if chiave != "anno"}
        crescita = percentuale(v["ricavi"] - ricavi_prima, ricavi_prima) if ricavi_prima else "n.d."
        pfn = v["debiti_finanziari"] - v["liquidita"]  # posizione finanziaria netta
        righe.append(
            f"Anno {a['anno']}: ricavi {v['ricavi']:.0f} mila euro (crescita {crescita}); "
            f"margine EBITDA {percentuale(v['ebitda'], v['ricavi'])}; "
            f"margine netto {percentuale(v['utile_netto'], v['ricavi'])}; "
            f"ROE {percentuale(v['utile_netto'], v['patrimonio_netto'])}; "
            f"liquidità corrente {rapporto(v['attivo_corrente'], v['passivo_corrente'])}; "
            f"debiti finanziari/patrimonio {rapporto(v['debiti_finanziari'], v['patrimonio_netto'])}; "
            f"posizione finanziaria netta {pfn:.0f} mila euro; "
            f"PFN/EBITDA {rapporto(pfn, v['ebitda'])}"
        )
        ricavi_prima = v["ricavi"]
    return "\n".join(righe)


def controlla_avviso(risultato: TaskOutput) -> tuple[bool, Any]:
    """Guardrail: la nota deve contenere l'avviso, altrimenti torna all'agente."""
    if AVVISO.lower() not in risultato.raw.lower():
        return (False, f"Manca l'avviso. Inizia la nota con la frase: {AVVISO}.")
    return (True, risultato.raw)


analista = Agent(
    role="Analista di bilancio per piccole imprese",
    goal="Descrivere con precisione come sono cambiati i conti di {azienda} negli anni del bilancio",
    backstory=(
        "Lavori da vent'anni con i bilanci delle PMI italiane. Non fai mai i conti a mente: "
        "usi il calcolatore e riporti i numeri esattamente come li ricevi."
    ),
    tools=[calcola_indici],
    max_iter=5,
    llm=MODELLO,
    verbose=True,
)

autore = Agent(
    role="Autore di note per la direzione",
    goal="Scrivere una nota breve ed equilibrata su rischi e opportunità di {azienda}",
    backstory=(
        "Scrivi per titolari che non sono esperti di finanza. Frasi corte, niente gergo, "
        "nessun consiglio di investimento: metti in fila i fatti e ciò che potrebbe andare bene o male."
    ),
    llm=MODELLO,
    verbose=True,
)

analisi = Task(
    description=(
        "Usa lo strumento Calcolatore di indici di bilancio sul file {file}.\n"
        "Riporta gli indici esattamente come li restituisce lo strumento: non ricalcolarli, "
        "non arrotondarli e non aggiungerne altri. Poi descrivi come sono cambiati negli anni."
    ),
    expected_output=(
        "Gli indici di ogni anno copiati dallo strumento, seguiti da 4-5 osservazioni "
        "sull'andamento, in italiano."
    ),
    agent=analista,
)

nota = Task(
    description=(
        f"Scrivi la nota su {{azienda}} usando solo i numeri dell'analisi. Inizia con la frase: {AVVISO}.\n"
        "Non inventare numeri e non dare consigli di acquisto, vendita o finanziamento."
    ),
    expected_output=(
        "Una nota in Markdown, in italiano, massimo 300 parole: l'avviso iniziale, un riassunto di 3 righe, "
        "3 rischi e 2 opportunità, ciascuno con il numero che lo giustifica."
    ),
    agent=autore,
    context=[analisi],
    guardrail=controlla_avviso,
    markdown=True,
    output_file="nota_bilancio.md",
)

crew = Crew(agents=[analista, autore], tasks=[analisi, nota], process=Process.sequential, verbose=True)

if __name__ == "__main__":
    if not Path(FILE_BILANCIO).exists():
        sys.exit(f"Non trovo {FILE_BILANCIO}: scaricalo e mettilo nella stessa cartella dello script.")

    risultato = crew.kickoff(inputs={"azienda": "Ferramenta Aurora S.r.l.", "file": FILE_BILANCIO})

    print("\n=== NOTA (salvata anche in nota_bilancio.md) ===")
    print(risultato.raw)
    # Gli stessi indici calcolati direttamente da Python: confrontali con i numeri della nota.
    print("\n=== INDICI CALCOLATI DA PYTHON, PER IL CONFRONTO ===")
    print(calcola_indici.run(nome_file=FILE_BILANCIO))

# /// script
# requires-python = ">=3.10,<3.14"
# dependencies = ["crewai[tools]==1.15.21"]
# ///
"""Un analista raccoglie notizie recenti su un'azienda quotata e un autore scrive una nota su rischi e opportunità."""
import os
from typing import Any

from crewai import Agent, Crew, Process, Task, TaskOutput
from crewai_tools import ScrapeWebsiteTool, SerperDevTool

# Il modello si sceglie nel file .env (capitolo 7). Se MODEL manca, si usa quello scritto qui.
MODELLO = os.getenv("MODEL", "openai/gpt-4.1-mini")
AVVISO = "Questa nota non è una consulenza finanziaria"


def controlla_nota(risultato: TaskOutput) -> tuple[bool, Any]:
    """Guardrail: senza avviso o senza link alle fonti, la nota torna all'agente."""
    testo = risultato.raw
    if AVVISO.lower() not in testo.lower():
        return (False, f"Manca l'avviso. Inizia la nota con la frase: {AVVISO}.")
    if "http" not in testo:
        return (False, "Mancano le fonti: accanto a ogni fatto metti il link dell'articolo da cui viene.")
    return (True, testo)


analista = Agent(
    role="Analista di notizie finanziarie",
    goal="Raccogliere i fatti recenti e verificabili su {azienda}, ognuno con data e link della fonte",
    backstory=(
        "Lavori nella redazione di un giornale economico. Distingui i fatti dalle opinioni "
        "e non riporti mai un numero senza il link da cui viene."
    ),
    tools=[SerperDevTool(search_type="news", n_results=8), ScrapeWebsiteTool()],
    inject_date=True,  # l'agente conosce la data di oggi e può capire cosa è "recente"
    max_iter=10,
    llm=MODELLO,
    verbose=True,
)

autore = Agent(
    role="Autore di note per risparmiatori prudenti",
    goal="Scrivere una nota breve ed equilibrata su rischi e opportunità di {azienda}",
    backstory=(
        "Scrivi per persone che non sono esperte di finanza. Non dai mai consigli di acquisto o vendita: "
        "metti in fila i fatti e spieghi cosa potrebbe andare bene o male."
    ),
    llm=MODELLO,
    verbose=True,
)

raccolta = Task(
    description=(
        "Cerca le notizie delle ultime settimane su {azienda}. Apri e leggi le 3 più rilevanti.\n"
        "Per ogni fatto annota la data, cosa è successo e il link. Se un numero compare in una sola fonte, "
        "segnalalo. Separa i fatti dalle opinioni dei giornalisti."
    ),
    expected_output="Un elenco di 5-8 fatti in italiano, ciascuno con data, descrizione breve e link della fonte.",
    agent=analista,
)

nota = Task(
    description=(
        f"Scrivi la nota su {{azienda}} usando solo i fatti raccolti. Inizia con la frase: {AVVISO}.\n"
        "Accanto a ogni fatto o numero metti il link della fonte. Non dare consigli di acquisto o vendita."
    ),
    expected_output=(
        "Una nota in Markdown, in italiano, massimo 350 parole: avviso iniziale, riassunto di 3 righe, "
        "3 rischi e 3 opportunità con link, e una sezione 'Da verificare' con i numeri da controllare "
        "sulle fonti ufficiali dell'azienda."
    ),
    agent=autore,
    context=[raccolta],
    guardrail=controlla_nota,
    markdown=True,
    output_file="nota_notizie.md",
)

crew = Crew(agents=[analista, autore], tasks=[raccolta, nota], process=Process.sequential, verbose=True)

if __name__ == "__main__":
    risultato = crew.kickoff(inputs={"azienda": "Enel"})
    print("\n=== NOTA (salvata anche in nota_notizie.md) ===")
    print(risultato)  # stampa il testo finale, lo stesso salvato nel file