131 lines
4.4 KiB
Python
131 lines
4.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
"""
|
|
Created on Wed Oct 22 15:15:02 2025
|
|
|
|
@author: Federico
|
|
"""
|
|
|
|
import os
|
|
from datetime import date, timedelta
|
|
|
|
import pandas as pd
|
|
import requests
|
|
|
|
# === Lettura token da file ===
|
|
with open("token.txt", "r") as f:
|
|
token = f.read().strip()
|
|
|
|
# === Parametri ===
|
|
isin = "DE000UP32UU0"
|
|
url = f"https://neo.ubs.com/api/ged-amc/external/report/v1/valuation/{isin}"
|
|
headers = {"Authorization": f"Bearer {token}"}
|
|
|
|
# === Intervallo management fee ===
|
|
management_from = date(2025, 11, 1)
|
|
management_to = date.today() - timedelta(days=1)
|
|
management_url = (
|
|
"https://neo.ubs.com/api/ged-amc/external/fee/v1/management/"
|
|
f"{isin}?fromDate={management_from:%Y-%m-%d}&toDate={management_to:%Y-%m-%d}"
|
|
)
|
|
performance_url = (
|
|
"https://neo.ubs.com/api/ged-amc/external/fee/v1/performance/"
|
|
f"{isin}?fromDate={management_from:%Y-%m-%d}&toDate={management_to:%Y-%m-%d}"
|
|
)
|
|
|
|
|
|
# Helper per trasformare il payload JSON in DataFrame pronti per Excel
|
|
def build_tables_from_payload(payload, base_name="MANAGEMENT_FEE"):
|
|
tables = {}
|
|
|
|
def normalize(obj):
|
|
return pd.json_normalize(obj, sep=".")
|
|
|
|
if isinstance(payload, list):
|
|
tables[base_name] = normalize(payload)
|
|
return tables
|
|
|
|
if isinstance(payload, dict):
|
|
if "content" in payload and isinstance(payload["content"], list):
|
|
tables[base_name] = normalize(payload["content"])
|
|
|
|
for key, val in payload.items():
|
|
if key == "content":
|
|
continue
|
|
if isinstance(val, list):
|
|
sheet_name = f"{base_name}_{key}".upper()[:31]
|
|
tables[sheet_name] = normalize(val)
|
|
|
|
if not tables:
|
|
tables[base_name] = normalize(payload)
|
|
|
|
return tables
|
|
|
|
|
|
# === Richiesta API valuation ===
|
|
response = requests.get(url, headers=headers)
|
|
print("Status code:", response.status_code)
|
|
|
|
if response.status_code == 200:
|
|
data = response.json()
|
|
|
|
# === Info generali del certificato ===
|
|
amc_info = data.get("amc", {})
|
|
print("\n--- INFO CERTIFICATO ---")
|
|
print("ISIN:", amc_info.get("isin"))
|
|
print("Descrizione:", amc_info.get("description"))
|
|
print("AUM:", amc_info.get("aum"))
|
|
|
|
# === Estraggo le varie asset class ===
|
|
constituents = data.get("constituents", {})
|
|
dfs = {k: pd.DataFrame(v) for k, v in constituents.items()}
|
|
|
|
# === Richiesta management fee ===
|
|
management_response = requests.get(management_url, headers=headers)
|
|
print("Management fee status code:", management_response.status_code)
|
|
management_tables = {}
|
|
|
|
if management_response.status_code == 200:
|
|
management_payload = management_response.json()
|
|
management_tables = build_tables_from_payload(management_payload)
|
|
else:
|
|
print("Errore nella richiesta management fee:", management_response.status_code)
|
|
print(management_response.text)
|
|
|
|
# === Richiesta performance fee ===
|
|
performance_response = requests.get(performance_url, headers=headers)
|
|
print("Performance fee status code:", performance_response.status_code)
|
|
performance_tables = {}
|
|
|
|
if performance_response.status_code == 200:
|
|
performance_payload = performance_response.json()
|
|
performance_tables = build_tables_from_payload(performance_payload, base_name="PERFORMANCE_FEE")
|
|
else:
|
|
print("Errore nella richiesta performance fee:", performance_response.status_code)
|
|
print(performance_response.text)
|
|
|
|
# === Salvataggio in Excel ===
|
|
output_file = f"Valuation_{isin}.xlsx"
|
|
with pd.ExcelWriter(output_file, engine="openpyxl") as writer:
|
|
# Salva anche info generali in un foglio
|
|
pd.DataFrame([amc_info]).to_excel(writer, sheet_name="AMC_INFO", index=False)
|
|
for name, df in dfs.items():
|
|
if not df.empty:
|
|
df.to_excel(writer, sheet_name=name[:31], index=False)
|
|
for name, df in management_tables.items():
|
|
if not df.empty:
|
|
df.to_excel(writer, sheet_name=name[:31], index=False)
|
|
for name, df in performance_tables.items():
|
|
if not df.empty:
|
|
df.to_excel(writer, sheet_name=name[:31], index=False)
|
|
|
|
print(f"\nFile salvato con successo: {os.path.abspath(output_file)}")
|
|
|
|
# Esempio: mostra i primi record di EQUITY
|
|
if "EQUITY" in dfs and not dfs["EQUITY"].empty:
|
|
print("\n--- EQUITY ---")
|
|
print(dfs["EQUITY"].head())
|
|
|
|
else:
|
|
print("Errore nella richiesta:", response.status_code)
|
|
print(response.text)
|