import warnings
import numpy as np
import pandas as pd
import yfinance as yf
warnings.filterwarnings("ignore")
# Univers borne : 15 grandes capitalisations US heterogenes (croissance, valeur,
# cycliques, sante, finance). Choix ex post documente dans la cellule precedente.
UNIVERS = ["AAPL", "MSFT", "INTC", "XOM", "CVX", "JPM",
"WMT", "KO", "PG", "F", "PFE", "CAT", "BA", "X", "T"]
def get(df, row, col):
# Lecture sure d'une cellule d'etat financier (NaN si ligne/colonne absente).
if df is None or row not in df.index or col not in df.columns:
return np.nan
v = df.loc[row, col]
try:
v = float(v)
except (TypeError, ValueError):
return np.nan
return v if pd.notna(v) else np.nan
def etats(tk, kind):
if kind == "annuel":
return tk.balance_sheet, tk.income_stmt, tk.cashflow
return tk.quarterly_balance_sheet, tk.quarterly_income_stmt, tk.quarterly_cashflow
def snapshot(tk, kind, col_now, col_prev):
# Valeurs brutes des lignes d'etat, pour [periode precedente, periode courante].
bs, inc, cf = etats(tk, kind)
vals = {}
for cle, (df_, row) in {
"TA": (bs, "Total Assets"),
"NI": (inc, "Net Income"),
"CFO": (cf, "Operating Cash Flow"),
"LTD": (bs, "Long Term Debt"),
"CA": (bs, "Current Assets"),
"CL": (bs, "Current Liabilities"),
"SHR": (bs, "Share Issued"),
"GP": (inc, "Gross Profit"),
"REV": (inc, "Total Revenue"),
}.items():
vals[cle] = [get(df_, row, c) for c in (col_prev, col_now)]
# Fallback dette long terme (certaines presentations fusionnent avec les leases).
if np.isnan(vals["LTD"]).all():
vals["LTD"] = [get(bs, "Long Term Debt And Capital Lease Obligation", c)
for c in (col_prev, col_now)]
return vals
def sous_scores(v):
# Les 9 sous-scores de Piotroski depuis les valeurs brutes [t-1, t].
# Retour None quand un input manque : le sous-score est compte hors score.
def ratio(num, den, i):
n, d = v[num][i], v[den][i]
if not (np.isfinite(n) and np.isfinite(d)) or d <= 0:
return np.nan
return n / d
roa = [ratio("NI", "TA", i) for i in (0, 1)]
cfo_ta = [ratio("CFO", "TA", i) for i in (0, 1)]
lev = [ratio("LTD", "TA", i) for i in (0, 1)]
liq = [ratio("CA", "CL", i) for i in (0, 1)]
marge = [ratio("GP", "REV", i) for i in (0, 1)]
rot = [ratio("REV", "TA", i) for i in (0, 1)]
def delta_pos(arr): # arr = [t-1, t] : le ratio s'est-il ameliore ?
a, b = arr
return None if not (np.isfinite(a) and np.isfinite(b)) else int(b > a)
def delta_inv(arr): # le ratio a-t-il diminue (levier) ?
a, b = arr
return None if not (np.isfinite(a) and np.isfinite(b)) else int(b < a)
s1 = None if not np.isfinite(roa[1]) else int(roa[1] > 0)
s2 = None if not np.isfinite(v["CFO"][1]) else int(v["CFO"][1] > 0)
s3 = delta_pos(roa)
s4 = None if not (np.isfinite(cfo_ta[1]) and np.isfinite(roa[1])) else int(cfo_ta[1] > roa[1])
s5 = delta_inv(lev)
s6 = delta_pos(liq)
s7 = None if not all(np.isfinite(x) for x in v["SHR"]) else int(v["SHR"][1] <= v["SHR"][0])
s8 = delta_pos(marge)
s9 = delta_pos(rot)
return [s1, s2, s3, s4, s5, s6, s7, s8, s9]
NOMS_SS = ["ROA>0", "CFO>0", "dROA", "Accr", "dLev", "dLiq", "dShr", "dMarge", "dRot"]
lignes = []
tickers = {}
for t in UNIVERS:
try:
tk = yf.Ticker(t)
cols = sorted(tk.balance_sheet.columns, reverse=True) # du plus recent au plus ancien
if len(cols) < 2:
print(f"{t}: moins de 2 exercices annuels disponibles, exclu")
continue
tickers[t] = tk
ss = sous_scores(snapshot(tk, "annuel", cols[0], cols[1]))
presents = [x for x in ss if x is not None]
lignes.append([t, str(cols[0].date()), *ss, sum(presents), len(presents)])
except Exception as e:
print(f"{t}: echec de recuperation ({type(e).__name__}), exclu honnetement")
df_annuel = pd.DataFrame(lignes, columns=["Ticker", "FY_t", *NOMS_SS, "F", "n_ss"])
print(f"Convention annuelle — {len(df_annuel)} titres (exercice le plus recent par ticker) :")
print(df_annuel.to_string(index=False))