sync/macro.py - master - SimpleSpace/AutoInvestDataSync

sync/macro.py
110 строк4 KB
Alex Just
Start commit for mvp 0.0.5 version
04 май 2026, 22:47
04 май 2026, 22:47e12863e
from __future__ import annotations

import json
import logging
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from urllib.request import urlopen
from urllib.error import URLError

logger = logging.getLogger(__name__)

API_URL = "https://statisticsoftheworld.com/api/v2/country/RUS"

EQMX_FIGI = "TCS00A101EJ5"
EQMX_NAME = "ВИМ – Индекс МосБиржи (EQMX)"


@dataclass
class MarketIndexData:
    name: str = EQMX_NAME
    current_price: float = 0.0
    price_year_ago: float = 0.0
    change_year_pct: float = 0.0
    current_date: str = ""
    year_ago_date: str = ""


@dataclass
class MacroData:
    inflation_imf_pct: float | None = None
    inflation_imf_year: str = ""
    inflation_cpi_wb_pct: float | None = None
    inflation_cpi_wb_year: str = ""
    gdp_growth_pct: float | None = None
    gdp_growth_year: str = ""
    lending_rate_pct: float | None = None
    deposit_rate_pct: float | None = None
    market_index: MarketIndexData = field(default_factory=MarketIndexData)


def fetch_macro_data() -> MacroData:
    macro = MacroData()
    try:
        with urlopen(API_URL, timeout=15) as resp:
            data = json.loads(resp.read())
    except (URLError, json.JSONDecodeError, Exception) as e:
        logger.warning("Failed to fetch macro data: %s", e)
        return macro

    indicators = {ind["id"]: ind for ind in data.get("indicators", [])}

    if "IMF.PCPIPCH" in indicators:
        ind = indicators["IMF.PCPIPCH"]
        macro.inflation_imf_pct = ind["value"]
        macro.inflation_imf_year = str(ind.get("year", ""))
    if "FP.CPI.TOTL.ZG" in indicators:
        ind = indicators["FP.CPI.TOTL.ZG"]
        macro.inflation_cpi_wb_pct = ind["value"]
        macro.inflation_cpi_wb_year = str(ind.get("year", ""))
    if "IMF.NGDP_RPCH" in indicators:
        ind = indicators["IMF.NGDP_RPCH"]
        macro.gdp_growth_pct = ind["value"]
        macro.gdp_growth_year = str(ind.get("year", ""))
    if "FR.INR.LEND" in indicators:
        macro.lending_rate_pct = indicators["FR.INR.LEND"]["value"]
    if "FR.INR.DPST" in indicators:
        macro.deposit_rate_pct = indicators["FR.INR.DPST"]["value"]

    logger.info("Macro: inflation(IMF %s)=%.1f%%, inflation(CPI WB %s)=%.1f%%",
                macro.inflation_imf_year, macro.inflation_imf_pct or 0,
                macro.inflation_cpi_wb_year, macro.inflation_cpi_wb_pct or 0)
    return macro


def fetch_market_index(tinvest) -> MarketIndexData:
    from sync.utils import quotation_to_float
    from t_tech.invest import CandleInterval

    idx = MarketIndexData()
    now = datetime.now(timezone.utc)
    from_ = now - timedelta(days=365)

    try:
        resp = tinvest.svc.market_data.get_candles(
            instrument_id=EQMX_FIGI,
            from_=from_,
            to=now,
            interval=CandleInterval.CANDLE_INTERVAL_WEEK,
        )
        candles = resp.candles
        if not candles:
            return idx

        first = candles[0]
        last = candles[-1]
        idx.current_price = quotation_to_float(last.close)
        idx.price_year_ago = quotation_to_float(first.open)
        idx.current_date = last.time.strftime("%Y-%m-%d")
        idx.year_ago_date = first.time.strftime("%Y-%m-%d")
        if idx.price_year_ago > 0:
            idx.change_year_pct = ((idx.current_price / idx.price_year_ago) - 1) * 100

        logger.info("Market index EQMX: %.2f (%s), year ago %.2f (%s), change %.2f%%",
                     idx.current_price, idx.current_date,
                     idx.price_year_ago, idx.year_ago_date,
                     idx.change_year_pct)
    except Exception as e:
        logger.warning("Failed to fetch market index: %s", e)

    return idx