def check_data()

in src/backend/domain/data_providers/tools/adj_prices_auto_checker/utils.py [0:0]

• 40 lines of code
• 8 McCabe index (conditional complexity)

def check_data(tickers: List[str], twlv_data: Dict[str, Any], alph_data: Dict[str, Any], volume_diff_treshold=0.03):
    for ticker in tickers:
        logging.info("Checking data for ticker: " + ticker)

        twlv_dict = {
            value["datetime"]: {
                "open": value["open"],
                "high": value["high"],
                "low": value["low"],
                "close": value["close"],
                "volume": value["volume"],
            }
            for value in twlv_data[ticker]["values"]
        }
        alph_dict = {
            value["datetime"]: {
                "open": value["open"],
                "high": value["high"],
                "low": value["low"],
                "close": value["close"],
                "volume": value["volume"],
            }
            for value in alph_data[ticker]["values"]
        }

        for datetime, alph_value in alph_dict.items():
            twlv_value = twlv_dict[datetime]

            if not all(
                [
                    round(float(alph_value[field]), 2) == round(float(twlv_value[field]), 2)
                    for field in ["open", "high", "low", "close"]
                ]
            ):
                warning_msg = f"Following prices at {datetime} dont match:\n" + build_price_table(twlv_value, alph_value)

                logging.warning(warning_msg)

            volume_diff = abs(float(alph_value["volume"]) - float(twlv_value["volume"]))
            percentage_diff = volume_diff / float(twlv_value["volume"])

            if percentage_diff > volume_diff_treshold:
                warning_msg = f"Following volumes in prices at {datetime} dont match:\n" + build_price_table(
                    twlv_value, alph_value
                )

                logging.warning(warning_msg)