import pandas as pd
import pytest

from freqtrade.constants import DEFAULT_TRADES_COLUMNS
from freqtrade.data.converter import populate_dataframe_with_trades
from freqtrade.data.converter.orderflow import (
    ORDERFLOW_ADDED_COLUMNS,
    stacked_imbalance,
    timeframe_to_DateOffset,
    trades_to_volumeprofile_with_total_delta_bid_ask,
)
from freqtrade.data.converter.trade_converter import trades_list_to_df
from freqtrade.data.dataprovider import DataProvider
from tests.strategy.strats.strategy_test_v3 import StrategyTestV3


BIN_SIZE_SCALE = 0.5


def read_csv(filename):
    converter_columns: list = ["side", "type"]
    return pd.read_csv(
        filename,
        skipinitialspace=True,
        index_col=0,
        parse_dates=True,
        date_format="ISO8601",
        converters={col: str.strip for col in converter_columns},
    )


@pytest.fixture
def populate_dataframe_with_trades_dataframe(testdatadir):
    return pd.read_feather(testdatadir / "orderflow/populate_dataframe_with_trades_DF.feather")


@pytest.fixture
def populate_dataframe_with_trades_trades(testdatadir):
    return pd.read_feather(testdatadir / "orderflow/populate_dataframe_with_trades_TRADES.feather")


@pytest.fixture
def public_trades_list(testdatadir):
    return read_csv(testdatadir / "orderflow/public_trades_list.csv").copy()


@pytest.fixture
def public_trades_list_simple(testdatadir):
    return read_csv(testdatadir / "orderflow/public_trades_list_simple_example.csv").copy()


def test_public_trades_columns_before_change(
    populate_dataframe_with_trades_dataframe, populate_dataframe_with_trades_trades
):
    assert populate_dataframe_with_trades_dataframe.columns.tolist() == [
        "date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]
    assert populate_dataframe_with_trades_trades.columns.tolist() == [
        "timestamp",
        "id",
        "type",
        "side",
        "price",
        "amount",
        "cost",
        "date",
    ]


def test_public_trades_mock_populate_dataframe_with_trades__check_orderflow(
    populate_dataframe_with_trades_dataframe, populate_dataframe_with_trades_trades
):
    """
    Tests the `populate_dataframe_with_trades` function's order flow calculation.

    This test checks the generated data frame and order flow for specific properties
    based on the provided configuration and sample data.
    """
    # Create copies of the input data to avoid modifying the originals
    dataframe = populate_dataframe_with_trades_dataframe.copy()
    trades = populate_dataframe_with_trades_trades.copy()
    # Convert the 'date' column to datetime format with milliseconds
    dataframe["date"] = pd.to_datetime(dataframe["date"], unit="ms")
    # Select the last rows and reset the index (optional, depends on usage)
    dataframe = dataframe.copy().tail().reset_index(drop=True)
    # Define the configuration for order flow calculation
    config = {
        "timeframe": "5m",
        "orderflow": {
            "cache_size": 1000,
            "max_candles": 1500,
            "scale": 0.005,
            "imbalance_volume": 0,
            "imbalance_ratio": 3,
            "stacked_imbalance_range": 3,
        },
    }
    # Apply the function to populate the data frame with order flow data
    df, _ = populate_dataframe_with_trades(None, config, dataframe, trades)
    # Extract results from the first row of the DataFrame
    results = df.iloc[0]
    t = results["trades"]
    of = results["orderflow"]

    # Assert basic properties of the results
    assert 0 != len(results)
    assert 151 == len(t)

    # --- Order Flow Analysis ---
    # Assert number of order flow data points
    assert 23 == len(of)  # Assert expected number of data points

    assert isinstance(of, dict)

    of_values = list(of.values())

    # Assert specific order flow values at the beginning of the DataFrame
    assert of_values[0] == {
        "bid": 0.0,
        "ask": 1.0,
        "delta": 4.999,
        "bid_amount": 0.0,
        "ask_amount": 4.999,
        "total_volume": 4.999,
        "total_trades": 1,
    }

    # Assert specific order flow values at the end of the DataFrame (excluding last row)
    assert of_values[-1] == {
        "bid": 0.0,
        "ask": 1.0,
        "delta": 0.103,
        "bid_amount": 0.0,
        "ask_amount": 0.103,
        "total_volume": 0.103,
        "total_trades": 1,
    }

    # Extract order flow from the last row of the DataFrame
    of = df.iloc[-1]["orderflow"]

    # Assert number of order flow data points in the last row
    assert 19 == len(of)  # Assert expected number of data points

    of_values1 = list(of.values())
    # Assert specific order flow values at the beginning of the last row
    assert of_values1[0] == {
        "bid": 1.0,
        "ask": 0.0,
        "delta": -12.536,
        "bid_amount": 12.536,
        "ask_amount": 0.0,
        "total_volume": 12.536,
        "total_trades": 1,
    }

    # Assert specific order flow values at the end of the last row
    assert pytest.approx(of_values1[-1]) == {
        "bid": 4.0,
        "ask": 3.0,
        "delta": -40.948,
        "bid_amount": 59.182,
        "ask_amount": 18.23399,
        "total_volume": 77.416,
        "total_trades": 7,
    }

    # --- Delta and Other Results ---

    # Assert delta value from the first row
    assert pytest.approx(results["delta"]) == -50.519
    # Assert min and max delta values from the first row
    assert results["min_delta"] == -79.469
    assert results["max_delta"] == 17.298

    # Assert that stacked imbalances are NaN (not applicable in this test)
    assert results["stacked_imbalances_bid"] == []
    assert results["stacked_imbalances_ask"] == []

    # Repeat assertions for the third from last row
    results = df.iloc[-2]
    assert pytest.approx(results["delta"]) == -20.862
    assert pytest.approx(results["min_delta"]) == -54.559999
    assert 82.842 == results["max_delta"]
    assert results["stacked_imbalances_bid"] == [234.97]
    assert results["stacked_imbalances_ask"] == [234.94]

    # Repeat assertions for the last row
    results = df.iloc[-1]
    assert pytest.approx(results["delta"]) == -49.302
    assert results["min_delta"] == -70.222
    assert pytest.approx(results["max_delta"]) == 11.213
    assert results["stacked_imbalances_bid"] == []
    assert results["stacked_imbalances_ask"] == []


def test_public_trades_trades_mock_populate_dataframe_with_trades__check_trades(
    populate_dataframe_with_trades_dataframe, populate_dataframe_with_trades_trades
):
    """
    Tests the `populate_dataframe_with_trades` function's handling of trades,
    ensuring correct integration of trades data into the generated DataFrame.
    """

    # Create copies of the input data to avoid modifying the originals
    dataframe = populate_dataframe_with_trades_dataframe.copy()
    trades = populate_dataframe_with_trades_trades.copy()

    # --- Data Preparation ---

    # Convert the 'date' column to datetime format with milliseconds
    dataframe["date"] = pd.to_datetime(dataframe["date"], unit="ms")

    # Select the final row of the DataFrame
    dataframe = dataframe.tail().reset_index(drop=True)

    # Filter trades to those occurring after or at the same time as the first DataFrame date
    trades = trades.loc[trades.date >= dataframe.date[0]]
    trades.reset_index(inplace=True, drop=True)  # Reset index for clarity

    # Assert the first trade ID to ensure filtering worked correctly
    assert trades["id"][0] == "313881442"

    # --- Configuration and Function Call ---

    # Define configuration for order flow calculation (used for context)
    config = {
        "timeframe": "5m",
        "orderflow": {
            "cache_size": 1000,
            "max_candles": 1500,
            "scale": 0.5,
            "imbalance_volume": 0,
            "imbalance_ratio": 3,
            "stacked_imbalance_range": 3,
        },
    }

    # Populate the DataFrame with trades and order flow data
    df, _ = populate_dataframe_with_trades(None, config, dataframe, trades)

    # --- DataFrame and Trade Data Validation ---

    row = df.iloc[0]  # Extract the first row for assertions

    # Assert DataFrame structure
    assert list(df.columns) == [
        # ... (list of expected column names)
        "date",
        "open",
        "high",
        "low",
        "close",
        "volume",
        "trades",
        "orderflow",
        "imbalances",
        "stacked_imbalances_bid",
        "stacked_imbalances_ask",
        "max_delta",
        "min_delta",
        "bid",
        "ask",
        "delta",
        "total_trades",
    ]
    # Assert delta, bid, and ask values
    assert pytest.approx(row["delta"]) == -50.519
    assert row["bid"] == 219.961
    assert row["ask"] == 169.442

    # Assert the number of trades
    assert len(row["trades"]) == 151

    # Assert specific details of the first trade
    t = row["trades"][0]
    assert list(t.keys()) == ["timestamp", "id", "type", "side", "price", "amount", "cost", "date"]
    assert trades["id"][0] == t["id"]
    assert int(trades["timestamp"][0]) == int(t["timestamp"])
    assert t["side"] == "sell"
    assert t["id"] == "313881442"
    assert t["price"] == 234.72


def test_public_trades_put_volume_profile_into_ohlcv_candles(public_trades_list_simple):
    """
    Tests the integration of volume profile data into OHLCV candles.

    This test verifies that
    the `trades_to_volumeprofile_with_total_delta_bid_ask`
    function correctly calculates the volume profile and that
    it correctly assigns the delta value from the volume profile to the
    corresponding candle in the `candles` DataFrame.
    """

    # Convert the trade list to a DataFrame
    trades_df = trades_list_to_df(public_trades_list_simple[DEFAULT_TRADES_COLUMNS].values.tolist())

    # Generate the volume profile with the specified bin size
    df = trades_to_volumeprofile_with_total_delta_bid_ask(trades_df, scale=BIN_SIZE_SCALE)

    # Assert the delta value in the total-bid/delta response of the second candle
    assert 0.14 == df.values.tolist()[1][2]

    # Alternative assertion using `.iat` accessor (assuming correct assignment logic)
    assert 0.14 == df["delta"].iat[1]


def test_public_trades_binned_big_sample_list(public_trades_list):
    """
    Tests the `trades_to_volumeprofile_with_total_delta_bid_ask` function
    with different bin sizes and verifies the generated DataFrame's structure and values.
    """

    # Define the bin size for the first test
    BIN_SIZE_SCALE = 0.05

    # Convert the trade list to a DataFrame
    trades = trades_list_to_df(public_trades_list[DEFAULT_TRADES_COLUMNS].values.tolist())

    # Generate the volume profile with the specified bin size
    df = trades_to_volumeprofile_with_total_delta_bid_ask(trades, scale=BIN_SIZE_SCALE)

    # Assert that the DataFrame has the expected columns
    assert df.columns.tolist() == [
        "bid",
        "ask",
        "delta",
        "bid_amount",
        "ask_amount",
        "total_volume",
        "total_trades",
    ]

    # Assert the number of rows in the DataFrame (expected 23 for this bin size)
    assert len(df) == 23

    # Assert that the index values are in ascending order and spaced correctly
    assert all(df.index[i] < df.index[i + 1] for i in range(len(df) - 1))
    assert df.index[0] + BIN_SIZE_SCALE == df.index[1]
    assert (trades["price"].min() - BIN_SIZE_SCALE) < df.index[0] < trades["price"].max()
    assert (df.index[0] + BIN_SIZE_SCALE) >= df.index[1]
    assert (trades["price"].max() - BIN_SIZE_SCALE) < df.index[-1] < trades["price"].max()

    # Assert specific values in the first and last rows of the DataFrame
    assert 32 == df["bid"].iloc[0]  # bid price
    assert 197.512 == df["bid_amount"].iloc[0]  # total bid amount
    assert 88.98 == df["ask_amount"].iloc[0]  # total ask amount
    assert 26 == df["ask"].iloc[0]  # ask price
    # delta (bid amount - ask amount)
    assert -108.532 == pytest.approx(df["delta"].iloc[0])

    assert 3 == df["bid"].iloc[-1]  # bid price
    assert 50.659 == df["bid_amount"].iloc[-1]  # total bid amount
    assert 108.21 == df["ask_amount"].iloc[-1]  # total ask amount
    assert 44 == df["ask"].iloc[-1]  # ask price
    assert 57.551 == df["delta"].iloc[-1]  # delta (bid amount - ask amount)

    # Repeat the process with a larger bin size
    BIN_SIZE_SCALE = 1

    # Generate the volume profile with the larger bin size
    df = trades_to_volumeprofile_with_total_delta_bid_ask(trades, scale=BIN_SIZE_SCALE)

    # Assert the number of rows in the DataFrame (expected 2 for this bin size)
    assert len(df) == 2

    # Repeat similar assertions for index ordering and spacing
    assert all(df.index[i] < df.index[i + 1] for i in range(len(df) - 1))
    assert (trades["price"].min() - BIN_SIZE_SCALE) < df.index[0] < trades["price"].max()
    assert (df.index[0] + BIN_SIZE_SCALE) >= df.index[1]
    assert (trades["price"].max() - BIN_SIZE_SCALE) < df.index[-1] < trades["price"].max()

    # Assert the value in the last row of the DataFrame with the larger bin size
    assert 1667.0 == df.index[-1]
    assert 710.98 == df["bid_amount"].iat[0]
    assert 111 == df["bid"].iat[0]
    assert 52.7199999 == pytest.approx(df["delta"].iat[0])  # delta


def test_public_trades_config_max_trades(
    default_conf, populate_dataframe_with_trades_dataframe, populate_dataframe_with_trades_trades
):
    dataframe = populate_dataframe_with_trades_dataframe.copy()
    trades = populate_dataframe_with_trades_trades.copy()
    default_conf["exchange"]["use_public_trades"] = True
    orderflow_config = {
        "timeframe": "5m",
        "orderflow": {
            "cache_size": 1000,
            "max_candles": 1,
            "scale": 0.005,
            "imbalance_volume": 0,
            "imbalance_ratio": 3,
            "stacked_imbalance_range": 3,
        },
    }

    df, _ = populate_dataframe_with_trades(None, default_conf | orderflow_config, dataframe, trades)
    assert df.delta.count() == 1


def test_public_trades_testdata_sanity(
    public_trades_list,
    public_trades_list_simple,
    populate_dataframe_with_trades_dataframe,
    populate_dataframe_with_trades_trades,
):
    assert 1000 == len(public_trades_list)
    assert 999 == len(populate_dataframe_with_trades_dataframe)
    assert 293532 == len(populate_dataframe_with_trades_trades)

    assert 7 == len(public_trades_list_simple)
    assert (
        5
        == public_trades_list_simple.loc[
            public_trades_list_simple["side"].str.contains("sell"), "id"
        ].count()
    )
    assert (
        2
        == public_trades_list_simple.loc[
            public_trades_list_simple["side"].str.contains("buy"), "id"
        ].count()
    )

    assert public_trades_list.columns.tolist() == [
        "timestamp",
        "id",
        "type",
        "side",
        "price",
        "amount",
        "cost",
        "date",
    ]

    assert public_trades_list.columns.tolist() == [
        "timestamp",
        "id",
        "type",
        "side",
        "price",
        "amount",
        "cost",
        "date",
    ]
    assert public_trades_list_simple.columns.tolist() == [
        "timestamp",
        "id",
        "type",
        "side",
        "price",
        "amount",
        "cost",
        "date",
    ]
    assert populate_dataframe_with_trades_dataframe.columns.tolist() == [
        "date",
        "open",
        "high",
        "low",
        "close",
        "volume",
    ]
    assert populate_dataframe_with_trades_trades.columns.tolist() == [
        "timestamp",
        "id",
        "type",
        "side",
        "price",
        "amount",
        "cost",
        "date",
    ]


def test_analyze_with_orderflow(
    default_conf_usdt,
    mocker,
    populate_dataframe_with_trades_dataframe,
    populate_dataframe_with_trades_trades,
):
    ohlcv_history = populate_dataframe_with_trades_dataframe
    # call without orderflow
    strategy = StrategyTestV3(config=default_conf_usdt)
    strategy.dp = DataProvider(default_conf_usdt, None, None)

    mocker.patch.object(strategy.dp, "trades", return_value=populate_dataframe_with_trades_trades)
    import freqtrade.data.converter.orderflow as orderflow_module

    spy = mocker.spy(orderflow_module, "trades_to_volumeprofile_with_total_delta_bid_ask")

    pair = "ETH/BTC"
    df = strategy.advise_indicators(ohlcv_history, {"pair:": pair})
    assert len(df) == len(ohlcv_history)
    assert "open" in df.columns
    assert spy.call_count == 0

    # Not expected to run - shouldn't have added orderflow columns
    for col in ORDERFLOW_ADDED_COLUMNS:
        assert col not in df.columns, f"Column {col} found in df.columns"

    default_conf_usdt["exchange"]["use_public_trades"] = True
    default_conf_usdt["orderflow"] = {
        "cache_size": 5,
        "max_candles": 5,
        "scale": 0.005,
        "imbalance_volume": 0,
        "imbalance_ratio": 3,
        "stacked_imbalance_range": 3,
    }

    strategy.config = default_conf_usdt
    # First round - builds cache
    df1 = strategy.advise_indicators(ohlcv_history, {"pair": pair})
    assert len(df1) == len(ohlcv_history)
    assert "open" in df1.columns
    assert spy.call_count == 5

    for col in ORDERFLOW_ADDED_COLUMNS:
        assert col in df1.columns, f"Column {col} not found in df.columns"

        if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"):
            assert df1[col].count() == 5, f"Column {col} has {df1[col].count()} non-NaN values"

    assert len(strategy._cached_grouped_trades_per_pair[pair]) == 5

    lastval_trades = df1.at[len(df1) - 1, "trades"]
    assert isinstance(lastval_trades, list)
    assert len(lastval_trades) == 122

    lastval_of = df1.at[len(df1) - 1, "orderflow"]
    assert isinstance(lastval_of, dict)

    spy.reset_mock()
    # Ensure caching works - call the same logic again.
    df2 = strategy.advise_indicators(ohlcv_history, {"pair": pair})
    assert len(df2) == len(ohlcv_history)
    assert "open" in df2.columns
    assert spy.call_count == 0
    for col in ORDERFLOW_ADDED_COLUMNS:
        assert col in df2.columns, f"Round2: Column {col} not found in df.columns"

        if col not in ("stacked_imbalances_bid", "stacked_imbalances_ask"):
            assert df2[col].count() == 5, (
                f"Round2: Column {col} has {df2[col].count()} non-NaN values"
            )

    lastval_trade2 = df2.at[len(df2) - 1, "trades"]
    assert isinstance(lastval_trade2, list)
    assert len(lastval_trade2) == 122

    lastval_of2 = df2.at[len(df2) - 1, "orderflow"]
    assert isinstance(lastval_of2, dict)


def test_stacked_imbalances_multiple_prices():
    """Test that stacked imbalances correctly returns multiple price levels when present"""
    # Test with empty result
    df_no_stacks = pd.DataFrame(
        {
            "bid_imbalance": [False, False, True, False],
            "ask_imbalance": [False, True, False, False],
        },
        index=[234.95, 234.96, 234.97, 234.98],
    )
    no_stacks = stacked_imbalance(df_no_stacks, "bid", stacked_imbalance_range=2)
    assert no_stacks == []

    # Create a sample DataFrame with known imbalances
    df = pd.DataFrame(
        {
            "bid_imbalance": [True, True, True, False, False, True, True, False, True],
            "ask_imbalance": [False, False, True, True, True, False, False, True, True],
        },
        index=[234.95, 234.96, 234.97, 234.98, 234.99, 235.00, 235.01, 235.02, 235.03],
    )
    # Test bid imbalances (should return prices in ascending order)
    bid_prices = stacked_imbalance(df, "bid", stacked_imbalance_range=2)
    assert bid_prices == [234.95, 234.96, 235.00]

    # Test ask imbalances (should return prices in descending order)
    ask_prices = stacked_imbalance(df, "ask", stacked_imbalance_range=2)
    assert ask_prices == [234.97, 234.98, 235.02]

    # Test with higher stacked_imbalance_range
    bid_prices_higher = stacked_imbalance(df, "bid", stacked_imbalance_range=3)
    assert bid_prices_higher == [234.95]


def test_timeframe_to_DateOffset():
    assert timeframe_to_DateOffset("1s") == pd.DateOffset(seconds=1)
    assert timeframe_to_DateOffset("1m") == pd.DateOffset(minutes=1)
    assert timeframe_to_DateOffset("5m") == pd.DateOffset(minutes=5)
    assert timeframe_to_DateOffset("1h") == pd.DateOffset(hours=1)
    assert timeframe_to_DateOffset("1d") == pd.DateOffset(days=1)
    assert timeframe_to_DateOffset("1w") == pd.DateOffset(weeks=1)
    assert timeframe_to_DateOffset("1M") == pd.DateOffset(months=1)
    assert timeframe_to_DateOffset("1y") == pd.DateOffset(years=1)
