mirror of
https://github.com/caty21/forex-dashboard.git
synced 2026-07-27 20:37:45 +00:00
Capucine Gest
345 lines
12 KiB
Python
345 lines
12 KiB
Python
"""
|
|
Scraper — Giuseppe Dellamotta / investinglive.com
|
|
Cible : articles "Rate hikes by year-end" publiés périodiquement (~mensuel)
|
|
Auteur : Claude (Anthropic) — usage personnel
|
|
|
|
Usage :
|
|
python investinglive_scraper.py # scrape les N derniers articles détectés
|
|
python investinglive_scraper.py --url <URL> # scrape un article spécifique
|
|
python investinglive_scraper.py --output json # sortie JSON (défaut : tableau console)
|
|
python investinglive_scraper.py --output json --save expectations.json
|
|
|
|
Stratégie de découverte d'articles :
|
|
1. Page auteur Giuseppe Dellamotta
|
|
2. Catégorie CentralBanks
|
|
3. Google Search comme fallback
|
|
Les articles contenant "Rate hikes by year-end" dans le corps sont sélectionnés.
|
|
"""
|
|
|
|
import requests
|
|
from bs4 import BeautifulSoup
|
|
import re
|
|
import json
|
|
import argparse
|
|
import time
|
|
import sys
|
|
from datetime import datetime
|
|
from urllib.parse import urljoin
|
|
|
|
# ── Configuration ──────────────────────────────────────────────────────────────
|
|
|
|
BASE_URL = "https://investinglive.com"
|
|
AUTHOR_URL = f"{BASE_URL}/author/giuseppe-dellamotta/"
|
|
CATEGORY_URL = f"{BASE_URL}/CentralBanks"
|
|
|
|
KEYWORD = "Rate hikes by year-end" # mot-clé discriminant
|
|
MAX_ARTICLES = 10 # nb max d'articles à scanner pour trouver les N derniers
|
|
|
|
# Headers navigateur réaliste (Chrome 124, Windows)
|
|
HEADERS = {
|
|
"User-Agent": (
|
|
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
|
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
|
"Chrome/124.0.0.0 Safari/537.36"
|
|
),
|
|
"Accept": (
|
|
"text/html,application/xhtml+xml,application/xml;"
|
|
"q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8"
|
|
),
|
|
"Accept-Language": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7",
|
|
"Accept-Encoding": "gzip, deflate, br",
|
|
"Connection": "keep-alive",
|
|
"Upgrade-Insecure-Requests": "1",
|
|
"Sec-Fetch-Dest": "document",
|
|
"Sec-Fetch-Mode": "navigate",
|
|
"Sec-Fetch-Site": "none",
|
|
"Sec-Fetch-User": "?1",
|
|
}
|
|
|
|
REQUEST_DELAY = 1.5 # secondes entre requêtes (politesse)
|
|
|
|
# ── Banques centrales reconnues ─────────────────────────────────────────────────
|
|
|
|
CB_NAMES = {
|
|
"fed": "Fed (USD)",
|
|
"ecb": "ECB (EUR)",
|
|
"boe": "BoE (GBP)",
|
|
"boj": "BoJ (JPY)",
|
|
"snb": "SNB (CHF)",
|
|
"boc": "BoC (CAD)",
|
|
"rba": "RBA (AUD)",
|
|
"rbnz": "RBNZ (NZD)",
|
|
}
|
|
|
|
def normalize_cb(raw: str) -> str:
|
|
key = raw.lower().strip(" *:")
|
|
for k, v in CB_NAMES.items():
|
|
if k in key:
|
|
return v
|
|
return raw.strip(" *:")
|
|
|
|
|
|
# ── HTTP helpers ────────────────────────────────────────────────────────────────
|
|
|
|
_session = requests.Session()
|
|
_session.headers.update(HEADERS)
|
|
|
|
def get_page(url: str, retries: int = 3) -> BeautifulSoup | None:
|
|
"""Fetch une page HTML et retourne un BeautifulSoup. None si échec."""
|
|
for attempt in range(retries):
|
|
try:
|
|
if attempt > 0:
|
|
time.sleep(REQUEST_DELAY * (attempt + 1))
|
|
resp = _session.get(url, timeout=20)
|
|
resp.raise_for_status()
|
|
return BeautifulSoup(resp.text, "html.parser")
|
|
except requests.HTTPError as e:
|
|
print(f" [HTTP {e.response.status_code}] {url}", file=sys.stderr)
|
|
if e.response.status_code in (403, 404):
|
|
return None # inutile de retenter
|
|
except requests.RequestException as e:
|
|
print(f" [Réseau] {e}", file=sys.stderr)
|
|
return None
|
|
|
|
|
|
# ── Extraction de la date depuis l'URL ─────────────────────────────────────────
|
|
|
|
def date_from_url(url: str) -> str:
|
|
"""Extrait YYYY-MM-DD depuis le slug YYYYMMDD en fin d'URL."""
|
|
m = re.search(r"(\d{8})/?$", url)
|
|
if m:
|
|
d = m.group(1)
|
|
return f"{d[:4]}-{d[4:6]}-{d[6:8]}"
|
|
return "?"
|
|
|
|
|
|
# ── Parsing du bloc de données ──────────────────────────────────────────────────
|
|
|
|
# Pattern : **RBNZ:** 76 bps (70% probability of no change at the next meeting)
|
|
_ENTRY_RE = re.compile(
|
|
r"\*?\*?\s*([A-Za-z/]+)\s*:+\s*\*?\*?\s*"
|
|
r"(\d+)\s*bps\s*"
|
|
r"\((\d+)%\s+probability\s+of\s+([^)]+)\)",
|
|
re.IGNORECASE,
|
|
)
|
|
|
|
def parse_rate_entries(text: str) -> list[dict]:
|
|
return [
|
|
{
|
|
"cb": normalize_cb(m.group(1)),
|
|
"bps": int(m.group(2)),
|
|
"prob_pct": int(m.group(3)),
|
|
"prob_desc": m.group(4).strip(),
|
|
}
|
|
for m in _ENTRY_RE.finditer(text)
|
|
]
|
|
|
|
|
|
def extract_rate_data(soup: BeautifulSoup, url: str) -> dict | None:
|
|
"""
|
|
Extrait les blocs 'Rate cuts by year-end' et 'Rate hikes by year-end'
|
|
depuis le HTML d'un article.
|
|
Retourne None si le keyword n'est pas trouvé.
|
|
"""
|
|
# Titre
|
|
h1 = soup.find("h1")
|
|
title = h1.get_text(strip=True) if h1 else "N/A"
|
|
|
|
# Texte complet de l'article
|
|
article = (
|
|
soup.find("article")
|
|
or soup.find("div", class_=re.compile(r"article|content|entry|post", re.I))
|
|
or soup.body
|
|
)
|
|
text = article.get_text(separator="\n") if article else soup.get_text(separator="\n")
|
|
|
|
# Vérifier la présence du keyword
|
|
if KEYWORD.lower() not in text.lower():
|
|
return None
|
|
|
|
# Parser les sections
|
|
result = {
|
|
"url": url,
|
|
"title": title,
|
|
"date": date_from_url(url),
|
|
"scraped_at": datetime.utcnow().isoformat() + "Z",
|
|
"rate_cuts": [],
|
|
"rate_hikes": [],
|
|
}
|
|
|
|
lines = [l.strip() for l in text.splitlines() if l.strip()]
|
|
current_section = None
|
|
|
|
for line in lines:
|
|
ll = line.lower()
|
|
if "rate cuts by year-end" in ll:
|
|
current_section = "cuts"
|
|
continue
|
|
if "rate hikes by year-end" in ll:
|
|
current_section = "hikes"
|
|
continue
|
|
|
|
if current_section and len(line) < 250:
|
|
entries = parse_rate_entries(line)
|
|
for e in entries:
|
|
if current_section == "cuts":
|
|
e["direction"] = "cut"
|
|
result["rate_cuts"].append(e)
|
|
else:
|
|
e["direction"] = "hike"
|
|
result["rate_hikes"].append(e)
|
|
|
|
# Stop si on sort manifestement du bloc (paragraphe long)
|
|
if current_section and len(line) > 300:
|
|
current_section = None
|
|
|
|
return result
|
|
|
|
|
|
# ── Découverte des articles ─────────────────────────────────────────────────────
|
|
|
|
def discover_article_urls(limit: int = MAX_ARTICLES) -> list[str]:
|
|
"""
|
|
Découvre les URLs d'articles de Giuseppe Dellamotta depuis :
|
|
1. Sa page auteur
|
|
2. La catégorie CentralBanks
|
|
Retourne une liste dédupliquée, triée du plus récent au plus ancien.
|
|
"""
|
|
found = []
|
|
|
|
# Source 1 : page auteur
|
|
soup = get_page(AUTHOR_URL)
|
|
if soup:
|
|
for a in soup.find_all("a", href=True):
|
|
href = a["href"]
|
|
# Articles de la catégorie centralbank avec slug YYYYMMDD
|
|
if "/centralbank/" in href and re.search(r"\d{8}/?$", href):
|
|
full = urljoin(BASE_URL, href)
|
|
if full not in found:
|
|
found.append(full)
|
|
time.sleep(REQUEST_DELAY)
|
|
|
|
# Source 2 : catégorie (si page auteur insuffisante)
|
|
if len(found) < limit:
|
|
soup2 = get_page(CATEGORY_URL)
|
|
if soup2:
|
|
for a in soup2.find_all("a", href=True):
|
|
href = a["href"]
|
|
if "/centralbank/" in href and re.search(r"\d{8}/?$", href):
|
|
full = urljoin(BASE_URL, href)
|
|
if full not in found:
|
|
found.append(full)
|
|
time.sleep(REQUEST_DELAY)
|
|
|
|
# Trier du plus récent (date dans l'URL) au plus ancien
|
|
def url_date_key(url):
|
|
m = re.search(r"(\d{8})/?$", url)
|
|
return m.group(1) if m else "00000000"
|
|
|
|
found.sort(key=url_date_key, reverse=True)
|
|
return found[:limit]
|
|
|
|
|
|
# ── Scraping principal ──────────────────────────────────────────────────────────
|
|
|
|
def scrape_recent(n: int = 2, specific_url: str = None) -> list[dict]:
|
|
"""
|
|
Scrape les n derniers articles contenant KEYWORD.
|
|
Si specific_url est fourni, scrape uniquement cet article.
|
|
"""
|
|
if specific_url:
|
|
urls_to_check = [specific_url]
|
|
else:
|
|
print(f"[Découverte] Recherche des articles de Giuseppe Dellamotta...", file=sys.stderr)
|
|
urls_to_check = discover_article_urls(limit=MAX_ARTICLES)
|
|
print(f" {len(urls_to_check)} articles candidats trouvés.", file=sys.stderr)
|
|
|
|
results = []
|
|
|
|
for url in urls_to_check:
|
|
if len(results) >= n and not specific_url:
|
|
break
|
|
|
|
print(f" → Scanning: {url}", file=sys.stderr)
|
|
soup = get_page(url)
|
|
if not soup:
|
|
continue
|
|
|
|
data = extract_rate_data(soup, url)
|
|
if data:
|
|
print(f" ✅ '{KEYWORD}' trouvé — {len(data['rate_hikes'])} BC hike / {len(data['rate_cuts'])} BC cut", file=sys.stderr)
|
|
results.append(data)
|
|
else:
|
|
print(f" ⏭ Keyword absent — article ignoré", file=sys.stderr)
|
|
|
|
time.sleep(REQUEST_DELAY)
|
|
|
|
return results
|
|
|
|
|
|
# ── Formatage console ───────────────────────────────────────────────────────────
|
|
|
|
def format_table(results: list[dict]) -> str:
|
|
lines = []
|
|
for r in results:
|
|
lines.append("=" * 65)
|
|
lines.append(f"📅 {r['date']} | {r['title'][:50]}")
|
|
lines.append(f"🔗 {r['url']}")
|
|
lines.append("")
|
|
|
|
all_entries = []
|
|
for e in r["rate_cuts"]:
|
|
all_entries.append((e["cb"], -e["bps"], e["prob_pct"], e["prob_desc"], "cut"))
|
|
for e in r["rate_hikes"]:
|
|
all_entries.append((e["cb"], e["bps"], e["prob_pct"], e["prob_desc"], "hike"))
|
|
|
|
# Trier par |bps| décroissant
|
|
all_entries.sort(key=lambda x: -abs(x[1]))
|
|
|
|
lines.append(f" {'Banque Centrale':<16} {'Attentes fin d\'année':>14} Prochain meeting")
|
|
lines.append(f" {'-'*16} {'-'*14} {'-'*30}")
|
|
for cb, bps, prob, desc, direction in all_entries:
|
|
arrow = "▲ +" if direction == "hike" else "▼ -"
|
|
lines.append(
|
|
f" {cb:<16} {arrow}{abs(bps):>3} bps {prob}% prob. of {desc}"
|
|
)
|
|
lines.append("")
|
|
|
|
lines.append("=" * 65)
|
|
lines.append(f"Scraped at: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
# ── CLI ─────────────────────────────────────────────────────────────────────────
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Scraper rate expectations — investinglive.com")
|
|
parser.add_argument("--url", help="URL spécifique à scraper (optionnel)")
|
|
parser.add_argument("--n", type=int, default=2, help="Nombre de derniers articles (défaut: 2)")
|
|
parser.add_argument("--output", choices=["table", "json"], default="table", help="Format de sortie")
|
|
parser.add_argument("--save", help="Fichier de sauvegarde JSON (optionnel)")
|
|
args = parser.parse_args()
|
|
|
|
results = scrape_recent(n=args.n, specific_url=args.url)
|
|
|
|
if not results:
|
|
print("❌ Aucun article trouvé avec le keyword.", file=sys.stderr)
|
|
sys.exit(1)
|
|
|
|
if args.output == "json" or args.save:
|
|
json_str = json.dumps(results, ensure_ascii=False, indent=2)
|
|
if args.save:
|
|
with open(args.save, "w", encoding="utf-8") as f:
|
|
f.write(json_str)
|
|
print(f"💾 Sauvegardé dans {args.save}", file=sys.stderr)
|
|
if args.output == "json":
|
|
print(json_str)
|
|
else:
|
|
print(format_table(results))
|
|
else:
|
|
print(format_table(results))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|