<reponame>globophobe/crypto-tick-data
import pandas as pd

from ..constants import VOLUME


def is_sample(data_frame, first_index, last_index):
    first_row = data_frame.loc[first_index]
    last_row = data_frame.loc[last_index]
    # For speed, short-circuit
    if first_row.timestamp == last_row.timestamp:
        if first_row.nanoseconds == last_row.nanoseconds:
            if first_row.tickRule == last_row.tickRule:
                if "symbol" in data_frame.columns:
                    if first_row.symbol == last_row.symbol:
                        return False
                else:
                    return False
    return True


def aggregate_trades(data_frame):
    df = data_frame.reset_index()
    idx = 0
    samples = []
    total_rows = len(df) - 1
    # Were there two or more trades?
    if len(df) > 1:
        for row in df.itertuples():
            index = row.Index
            last_index = index - 1
            if index > 0:
                is_last_iteration = index == total_rows
                # Is this the last iteration?
                if is_last_iteration:
                    # If equal, one sample
                    if not is_sample(df, idx, index):
                        # Aggregate from idx to end of data frame
                        sample = df.loc[idx:]
                        samples.append(agg_trades(sample))
                    # Otherwise, two samples.
                    else:
                        # Aggregate from idx to last_index
                        sample = df.loc[idx:last_index]
                        samples.append(agg_trades(sample))
                        # Append last row.
                        sample = df.loc[index:]
                        assert len(sample) == 1
                        samples.append(agg_trades(sample))
                # Is the last row equal to the current row?
                elif is_sample(df, last_index, index):
                    # Aggregate from idx to last_index
                    sample = df.loc[idx:last_index]
                    aggregated_sample = agg_trades(sample)
                    samples.append(aggregated_sample)
                    idx = index
    # Only one trade in data_frame
    elif len(df) == 1:
        aggregated_sample = agg_trades(df)
        samples.append(aggregated_sample)
    return pd.DataFrame(samples)


def agg_trades(data_frame):
    last_row = data_frame.iloc[-1]
    timestamp = last_row.timestamp
    last_price = last_row.price
    ticks = len(data_frame)
    # Is there more than 1 trade to aggregate?
    if ticks > 1:
        volume = data_frame.volume.sum()
        notional = data_frame.notional.sum()
    else:
        volume = last_row.volume
        notional = last_row.notional
    data = {
        "timestamp": timestamp,
        "nanoseconds": last_row.nanoseconds,
        "price": last_price,
        "volume": volume,
        "notional": notional,
        "ticks": ticks,
        "tickRule": last_row.tickRule,
    }
    if "symbol" in data_frame.columns:
        data.update({"symbol": last_row.symbol})
    return data


def aggregate_filter(data_frame, attr=VOLUME, value=1000):
    samples = []
    total_rows = len(data_frame)
    if len(data_frame):
        next_index = 0
        for row in data_frame.itertuples():
            index = row.Index
            if getattr(row, attr) >= value:
                if index == 0:
                    sample = data_frame.loc[:index]
                else:
                    sample = data_frame.loc[next_index:index]
                samples.append(agg_filter(sample, is_min_val=True))
                next_index = index + 1
        if next_index < total_rows:
            sample = data_frame.loc[next_index:]
            samples.append(agg_filter(sample))
    return samples


def agg_filter(df, is_min_val=False):
    data = {}
    if is_min_val:
        last_row = df.iloc[-1]
        data.update(
            {
                "timestamp": last_row.timestamp,
                "price": last_row.price,
                "volume": last_row.volume,
                "notional": last_row.notional,
                "tickRule": last_row.tickRule,
                "ticks": last_row.ticks,
            }
        )
    if not is_min_val or len(df) > 1:
        buy_df = df[df.tickRule == 1]
        data.update(
            {
                "high": df.price.max(),
                "low": df.price.min(),
                "totalBuyVolume": buy_df.volume.sum(),
                "totalVolume": df.volume.sum(),
                "totalBuyNotional": buy_df.notional.sum(),
                "totalNotional": df.notional.sum(),
                "totalBuyTicks": buy_df.ticks.sum(),
                "totalTicks": df.ticks.sum(),
            }
        )
    return data

import pandas as pd

from ..constants import VOLUME


def is_sample(data_frame, first_index, last_index):
    first_row = data_frame.loc[first_index]
    last_row = data_frame.loc[last_index]
    # For speed, short-circuit
    if first_row.timestamp == last_row.timestamp:
        if first_row.nanoseconds == last_row.nanoseconds:
            if first_row.tickRule == last_row.tickRule:
                if "symbol" in data_frame.columns:
                    if first_row.symbol == last_row.symbol:
                        return False
                else:
                    return False
    return True


def aggregate_trades(data_frame):
    df = data_frame.reset_index()
    idx = 0
    samples = []
    total_rows = len(df) - 1
    # Were there two or more trades?
    if len(df) > 1:
        for row in df.itertuples():
            index = row.Index
            last_index = index - 1
            if index > 0:
                is_last_iteration = index == total_rows
                # Is this the last iteration?
                if is_last_iteration:
                    # If equal, one sample
                    if not is_sample(df, idx, index):
                        # Aggregate from idx to end of data frame
                        sample = df.loc[idx:]
                        samples.append(agg_trades(sample))
                    # Otherwise, two samples.
                    else:
                        # Aggregate from idx to last_index
                        sample = df.loc[idx:last_index]
                        samples.append(agg_trades(sample))
                        # Append last row.
                        sample = df.loc[index:]
                        assert len(sample) == 1
                        samples.append(agg_trades(sample))
                # Is the last row equal to the current row?
                elif is_sample(df, last_index, index):
                    # Aggregate from idx to last_index
                    sample = df.loc[idx:last_index]
                    aggregated_sample = agg_trades(sample)
                    samples.append(aggregated_sample)
                    idx = index
    # Only one trade in data_frame
    elif len(df) == 1:
        aggregated_sample = agg_trades(df)
        samples.append(aggregated_sample)
    return pd.DataFrame(samples)


def agg_trades(data_frame):
    last_row = data_frame.iloc[-1]
    timestamp = last_row.timestamp
    last_price = last_row.price
    ticks = len(data_frame)
    # Is there more than 1 trade to aggregate?
    if ticks > 1:
        volume = data_frame.volume.sum()
        notional = data_frame.notional.sum()
    else:
        volume = last_row.volume
        notional = last_row.notional
    data = {
        "timestamp": timestamp,
        "nanoseconds": last_row.nanoseconds,
        "price": last_price,
        "volume": volume,
        "notional": notional,
        "ticks": ticks,
        "tickRule": last_row.tickRule,
    }
    if "symbol" in data_frame.columns:
        data.update({"symbol": last_row.symbol})
    return data


def aggregate_filter(data_frame, attr=VOLUME, value=1000):
    samples = []
    total_rows = len(data_frame)
    if len(data_frame):
        next_index = 0
        for row in data_frame.itertuples():
            index = row.Index
            if getattr(row, attr) >= value:
                if index == 0:
                    sample = data_frame.loc[:index]
                else:
                    sample = data_frame.loc[next_index:index]
                samples.append(agg_filter(sample, is_min_val=True))
                next_index = index + 1
        if next_index < total_rows:
            sample = data_frame.loc[next_index:]
            samples.append(agg_filter(sample))
    return samples


def agg_filter(df, is_min_val=False):
    data = {}
    if is_min_val:
        last_row = df.iloc[-1]
        data.update(
            {
                "timestamp": last_row.timestamp,
                "price": last_row.price,
                "volume": last_row.volume,
                "notional": last_row.notional,
                "tickRule": last_row.tickRule,
                "ticks": last_row.ticks,
            }
        )
    if not is_min_val or len(df) > 1:
        buy_df = df[df.tickRule == 1]
        data.update(
            {
                "high": df.price.max(),
                "low": df.price.min(),
                "totalBuyVolume": buy_df.volume.sum(),
                "totalVolume": df.volume.sum(),
                "totalBuyNotional": buy_df.notional.sum(),
                "totalNotional": df.notional.sum(),
                "totalBuyTicks": buy_df.ticks.sum(),
                "totalTicks": df.ticks.sum(),
            }
        )
    return data
# For speed, short-circuit # Were there two or more trades? # Is this the last iteration? # If equal, one sample # Aggregate from idx to end of data frame # Otherwise, two samples. # Aggregate from idx to last_index # Append last row. # Is the last row equal to the current row? # Aggregate from idx to last_index # Only one trade in data_frame # Is there more than 1 trade to aggregate?