Portfolio_Risk_Management/tests/test_data.py

import numpy as np
import pandas as pd
import pytest

from quantcore.data.download_service import splice_adjusted
from quantcore.data.fx import FXConverter
from quantcore.data.yahoo import apply_price_scale, normalise_info
from quantcore.universe.sources import parse_ishares_holdings
from quantcore.universe.symbols import normalise_share_class_name, to_yahoo_symbol


def test_fx_conversion_and_log_return_identity():
    idx = pd.bdate_range("2024-01-01", periods=5)
    fx = pd.DataFrame({"USD": [0.85, 0.86, 0.84, 0.85, 0.87], "CHF": 1.0}, index=idx)
    conv = FXConverter(fx, base="CHF")
    px = pd.DataFrame({"A": [10.0, 11.0, 10.5, 10.8, 11.2], "B": [100.0, 101.0, 99.0, 98.0, 100.0]}, index=idx)
    cur = pd.Series({"A": "USD", "B": "CHF"})
    chf = conv.convert_prices(px, cur)
    np.testing.assert_allclose(chf["A"], px["A"] * fx["USD"])
    np.testing.assert_allclose(chf["B"], px["B"])
    # CHF log-return = local log-return + FX log-return
    r_chf = np.log(chf).diff().iloc[1:]
    r_loc = np.log(px).diff().iloc[1:]
    r_fx = conv.fx_log_returns(cur, idx).iloc[1:]
    np.testing.assert_allclose(r_chf.values, (r_loc + r_fx).values)
    with pytest.raises(KeyError):
        conv.convert_prices(px, pd.Series({"A": "JPY", "B": "CHF"}))


def test_gbp_pence_normalisation():
    info = normalise_info("HSBA.L", {"currency": "GBp", "marketCap": 1e11, "regularMarketPrice": 1533.8,
                                      "firstTradeDateMilliseconds": 584089200000, "quoteType": "EQUITY"})
    assert info["currency"] == "GBP" and info["price_scale"] == 100.0
    assert info["last_price"] == pytest.approx(15.338)
    assert info["first_trade_date"] == "1988-07-05"
    px = pd.DataFrame({"HSBA.L": [1500.0, 1600.0], "NESN.SW": [80.0, 81.0]})
    scaled = apply_price_scale(px, pd.Series({"HSBA.L": 100.0}))
    assert scaled["HSBA.L"].iloc[0] == 15.0 and scaled["NESN.SW"].iloc[1] == 81.0


def test_splice_rescales_history_on_dividend():
    idx = pd.bdate_range("2024-01-01", periods=10)
    stored = pd.DataFrame({"A": np.linspace(100, 109, 10), "B": np.linspace(50, 59, 10)}, index=idx)
    fresh_idx = pd.bdate_range("2024-01-08", periods=7)  # overlaps the last 5 stored days, adds 2 new
    fresh = pd.DataFrame({"A": np.concatenate([stored["A"].iloc[5:].values * 0.98, [110 * 0.98, 111 * 0.98]]),
                          "B": np.concatenate([stored["B"].iloc[5:].values, [60.0, 61.0]])}, index=fresh_idx)
    out, rescaled = splice_adjusted(stored, fresh)
    assert rescaled == ["A"]
    assert len(out) == 12
    np.testing.assert_allclose(out["A"].iloc[:5], stored["A"].iloc[:5] * 0.98)  # history rescaled
    np.testing.assert_allclose(out["A"].iloc[5:], fresh["A"])  # fresh window authoritative
    np.testing.assert_allclose(out["B"], np.concatenate([stored["B"].values, [60.0, 61.0]]))


def test_symbol_mapping_rules():
    assert to_yahoo_symbol("NOVO B", "Omx Nordic Exchange Copenhagen A/S") == "NOVO-B.CO"
    assert to_yahoo_symbol("BT.A", "London Stock Exchange") == "BT-A.L"
    assert to_yahoo_symbol("RR.", "London Stock Exchange") == "RR.L"
    assert to_yahoo_symbol("NESN", "SIX Swiss Exchange") == "NESN.SW"
    assert to_yahoo_symbol("BRK.B", country="United States") == "BRK-B"
    assert to_yahoo_symbol("NDA FI", "Nasdaq Omx Helsinki Ltd.") == "NDA-FI.HE"
    with pytest.raises(ValueError):
        to_yahoo_symbol("X", "Mars Exchange")
    assert normalise_share_class_name("Alphabet Inc. Class A") == normalise_share_class_name("Alphabet Inc. Class C")
    assert normalise_share_class_name("Investor AB ser. B") == normalise_share_class_name("Investor Class A")


def test_parse_ishares_holdings():
    text = ('Fondsposition per,"18.Sept.2026"\n \nEmittententicker,Name,Sektor,Anlageklasse,Marktwert,'
            'Gewichtung (%),Nominalwert,Nominale,Kurs,Standort,Börse,Marktwährung\n'
            '"ASML","ASML HOLDING","IT","Aktien","438.829.188,00","4,23","438.829.188,00","303.478,00","1.446,00",'
            '"Niederlande","Euronext Amsterdam","EUR"\n'
            '"NOVO B","NOVO NORDISK CLASS B","Gesundheitsversorgung","Aktien","1,00","1,00","1,00","1,00","1,00",'
            '"Dänemark","Omx Nordic Exchange Copenhagen A/S","DKK"\n'
            '"XEUR","EUR CASH","Barmittel und/oder Derivate","Geldmarkt","1,00","0,10","1,00","1,00","1,00",'
            '"Deutschland","-","EUR"\n')
    df = parse_ishares_holdings(text)
    assert list(df["symbol"]) == ["ASML.AS", "NOVO-B.CO"]
    assert df.iloc[0]["gics_sector"] == "Information Technology" and df.iloc[0]["country"] == "Netherlands"
    assert df.iloc[0]["index_weight"] == pytest.approx(4.23)
    assert df.iloc[0]["source_date"] == "2026-09-18"


# -- candidate sources: which list a setup builds its universe from ---------------------------------
def _frame(symbols, source):
    from quantcore.universe.sources import CANDIDATE_COLUMNS

    rows = [{c: "" for c in CANDIDATE_COLUMNS} | {"symbol": s, "country": "Switzerland", "region": "Europe",
                                                 "source": source, "source_date": "2026-10-06"} for s in symbols]
    return pd.DataFrame(rows, columns=CANDIDATE_COLUMNS)


def test_europe_falls_back_to_the_shipped_list_not_to_wikipedia(monkeypatch):
    """iShares down (it answered 404 in October 2026): the shipped snapshot, with exact symbols, is used.
    Wikipedia's table yields guesses such as BNPP.PA or NOV-N.SW that Yahoo does not know."""
    from quantcore.universe import sources as S

    calls = []
    monkeypatch.setattr(S, "fetch_stoxx600_ishares", lambda timeout=30.0: (_ for _ in ()).throw(RuntimeError("404")))
    monkeypatch.setattr(S, "fetch_stoxx600_wikipedia", lambda timeout=30.0: calls.append("wiki") or _frame(["X"] * 200, "w"))
    monkeypatch.setattr(S, "save_static_candidates", lambda *a, **k: calls.append("save"))
    df = S.fetch_candidates("Europe")
    assert calls == [] and len(df) > 500 and set(df["source"]) == {"ishares_stoxx600"}
    assert {"BNP.PA", "NOVN.SW", "VWS.CO", "UCG.MI"} <= set(df["symbol"])


def test_offline_setup_uses_the_shipped_lists_without_the_web(monkeypatch):
    from quantcore.universe import sources as S

    def no_web(*a, **k):
        raise AssertionError("no web access with allow_web=False")

    monkeypatch.setattr(S, "_get", no_web)
    assert len(S.fetch_candidates("US", allow_web=False)) > 400
    assert len(S.fetch_candidates("Europe", allow_web=False)) > 500


def test_ishares_tries_the_next_address(monkeypatch):
    from quantcore.universe import sources as S

    tried = []

    def fake_get(url, timeout=30.0):
        tried.append(url)
        if len(tried) == 1:
            raise RuntimeError("404 Client Error")
        return b"csv"

    monkeypatch.setattr(S, "_get", fake_get)
    monkeypatch.setattr(S, "parse_ishares_holdings", lambda raw: _frame([f"S{i}.SW" for i in range(150)], "ishares"))
    df = S.fetch_stoxx600_ishares()
    assert tried == list(S.ISHARES_STOXX600_URLS[:2]) and len(df) == 150


def test_unknown_symbols_are_summarised_not_logged_one_by_one(monkeypatch, caplog):
    import logging

    from quantcore.config import Settings
    from quantcore.data.yahoo import YahooClient

    client = YahooClient(Settings(yahoo_info_pause=0.0))
    known = {"quoteType": "EQUITY", "currency": "CHF", "longName": "Known AG", "exchange": "EBS"}
    monkeypatch.setattr(YahooClient, "_info_fields",
                        lambda self, s: logging.getLogger("yfinance").error("HTTP Error 404 %s", s) or
                        ({} if s.startswith("GONE") else known))
    with caplog.at_level(logging.INFO):
        df = client.ticker_infos(["KNOWN.SW", "GONE1.PA", "GONE2.SW"])
    assert list(df.index) == ["KNOWN.SW", "GONE1.PA", "GONE2.SW"] and df.loc["KNOWN.SW", "currency"] == "CHF"
    assert not [r for r in caplog.records if r.name == "yfinance"]          # no per-symbol ERROR lines
    summary = [r.getMessage() for r in caplog.records if "no quote for 2 of 3" in r.getMessage()]
    assert summary and "GONE1.PA, GONE2.SW" in summary[0]
    assert logging.getLogger("yfinance").level == logging.NOTSET           # restored afterwards