Configure option for zero volume

[robcarver17]

[robcarver17]

Let's make some notes about price capture behaviour.

update_historical_prices

    # Get *intraday* data (defaults to hourly)
    result = get_and_add_prices_for_frequency(
        data, contract_object, frequency=intraday_frequency
    )
    if result is failure:
        # Skip daily data if intraday not working

        return None

    # Get daily data
    # we don't care about the result flag for this
    get_and_add_prices_for_frequency(data, contract_object, frequency=daily_frequency)

The reason for this is that if can't get intraday then we will lose the intraday data forever, as we don't allow ourselves to add data older than the last time stamp in a given dataset.

This assumes that the only reason for not getting prices is a temporary IB issue.

But for things with crap price history we might never get intraday data.

Make this configurable? Can we distinguish between 'no data' and 'IB issue' better?

[robcarver17]

    if len(broker_prices) == 0:
        data.log.msg("No prices from broker for %s" % str(contract_object))
        return failure

No way to distinguish between no data and failure

[robcarver17]

    error_or_rows_added = db_futures_prices.update_prices_for_contract(
        contract_object, broker_prices, check_for_spike=True
    )

    if error_or_rows_added is spike_in_data:
        report_price_spike(data, contract_object)
        return failure

[robcarver17]

interactive_manual_check_historical_prices

    intraday_frequency = diag_prices.get_intraday_frequency_for_historical_download()
    daily_frequency = DAILY_PRICE_FREQ

    get_and_check_prices_for_frequency(
        data, contract_object, frequency=intraday_frequency
    )
    get_and_check_prices_for_frequency(data, contract_object, frequency=daily_frequency)

Always does daily.

[robcarver17]

    broker_prices = broker_data.get_prices_at_frequency_for_contract_object(
        contract_object, frequency
    )
    if len(broker_prices) == 0:
        print("No broker prices found for %s nothing to check" % str(contract_object))
        return failure

    print("\n\n Manually checking prices for %s \n\n" % str(contract_object))
    new_prices_checked = manual_price_checker(
        old_prices,
        broker_prices,
        column_to_check="FINAL",
        delta_columns=["OPEN", "HIGH", "LOW"],
        type_new_data=futuresContractPrices,
    )
    price_updater.update_prices_for_contract(
        contract_object, new_prices_checked, check_for_spike=False
    )

Updater without spike checking.

[robcarver17]

sysdata/futures/futures_per_contract_prices.py

    def update_prices_for_contract(
        self,
        contract_object: futuresContract,
        new_futures_per_contract_prices: futuresContractPrices,
        check_for_spike: bool = True,
    ) -> int:
        """
        Reads existing data, merges with new_futures_prices, writes merged data

        :param new_futures_prices:
        :return: int, number of rows
        """
        new_log = contract_object.log(self.log)

        if len(new_futures_per_contract_prices) == 0:
            new_log.msg("No new data")
            return 0

        old_prices = self.get_prices_for_contract_object(contract_object)
        merged_prices = old_prices.add_rows_to_existing_data(
            new_futures_per_contract_prices, check_for_spike=check_for_spike
        )

        if merged_prices is spike_in_data:
            new_log.msg(
                "Price has moved too much - will need to manually check - no price update done"
            )
            return spike_in_data

        rows_added = len(merged_prices) - len(old_prices)

        if rows_added < 0:
            new_log.critical("Can't remove prices something gone wrong!")
            return 0

        elif rows_added == 0:
            if len(old_prices) == 0:
                new_log.msg("No existing or additional data")
                return 0
            else:
                new_log.msg("No additional data since %s " % str(old_prices.index[-1]))
            return 0

        # We have guaranteed no duplication
        self.write_prices_for_contract_object(
            contract_object, merged_prices, ignore_duplication=True
        )

        new_log.msg("Added %d additional rows of data" % rows_added)

        return rows_added

Returns number of rows - could return success or failure mode instead?

Spike checking done inside price object.

[robcarver17]

sysobjects/futures_per_contract_prices.py

    def add_rows_to_existing_data(
        self, new_futures_per_contract_prices, check_for_spike=True
    ):
        """
        Merges self with new data.
        Only newer data will be added

        :param new_futures_per_contract_prices: another futures per contract prices object

        :return: merged futures_per_contract object
        """

        merged_futures_prices = merge_newer_data(
            pd.DataFrame(self),
            new_futures_per_contract_prices,
            check_for_spike=check_for_spike,
            column_to_check=FINAL_COLUMN,
        )

        if merged_futures_prices is spike_in_data:
            return spike_in_data

        merged_futures_prices = futuresContractPrices(merged_futures_prices)

        return merged_futures_prices

[robcarver17]

syscore/merge_data.py

def merge_newer_data(
    old_data, new_data, check_for_spike=True, column_to_check=arg_not_supplied
):
    """
    Merge new data, with old data. Any new data that is older than the newest old data will be ignored

    If check_for_spike will return data_error if price moves too much on join point

    :param old_data: pd.Series or DataFrame
    :param new_data: pd.Series or DataFrame
    :param check_for_spike: bool
    :param column_to_check: column name to check for spike

    :return:  pd.Series or DataFrame
    """
    merged_data_with_status = merge_newer_data_no_checks(old_data, new_data)

    # check for spike
    if check_for_spike:
        merged_data_with_status = spike_check_merged_data(
            merged_data_with_status,
            column_to_check=column_to_check,
        )
        if merged_data_with_status.spike_present:
            return spike_in_data

    merged_data = merged_data_with_status.merged_data

    return merged_data

[robcarver17]

sysbrokers/IB/ib_futures_contract_price_data.py

        ## It's important that the data is in local time zone so that this works
        price_data = price_data.remove_future_data()

        ## Ignore zeros if no volumes (if volume could be real price eg crude oil)
        price_data = price_data.remove_zero_prices_if_zero_volumes()

[robcarver17]

Workflow should be something like this:

• get prices from IB - apply no filtering or zero volume type stuff at this stage DONE
• return failure mode not zero length to distinguish problems DONE
• process those prices in some configurable way - applying past/future filtering and zero volumes DONE
• option to change configuration for cleaning in interactive mode DONE
• apply spike checking in some configrable way (does this really need changing? This is sort of a seperate thing) NO CHANGE REQUIRED
• spike size is pulled directly into syscore - this feels wrong - should be passed down from top level DONE
• if hourly fails then configurable or not whether we get daily. If hourly zero length then we do get daily DONE
• fair amount of overlapping code in interactive versus automated get prices DONE

[bug-or-feature]

Maybe make the intraday resolution configurable per instrument?

[robcarver17]

A bit of a pain, and as liquidity changes over time you would need to keep updating the config. I think not.

[robcarver17]

Let's start by moving the IB filtering stuff out of the broker code.

 #### use this function instead
    def get_cleaned_prices_at_frequency_for_contract_object(
        self, contract_object: futuresContract, frequency: Frequency,
            cleaning_control_args_override = arg_not_supplied
    ) -> futuresContractPrices:

        broker_prices_raw = \
                self.get_prices_at_frequency_for_contract_object(contract_object=contract_object,
                                                         frequency = frequency)

        if broker_prices_raw is failure:
            return failure

        # FIXME NEEDS TO BE CONFIGURABLE AND INSIDE SOME FUNCTION
        ## It's important that the data is in local time zone so that this works
        broker_prices_no_future = broker_prices_raw.remove_future_data()

        ## Ignore zeros if no volumes (if volume could be real price eg crude oil)
        broker_prices = broker_prices_no_future.remove_zero_prices_if_zero_volumes()
        # END FIXME

        return broker_prices

    def get_prices_at_frequency_for_contract_object(
        self, contract_object: futuresContract, frequency: Frequency
    ) -> futuresContractPrices:

        return self.broker_futures_contract_price_data.get_prices_at_frequency_for_contract_object(
            contract_object, frequency
        )

[robcarver17]

Some new config elements

# Behaviour of price filtering
ignore_future_prices: True
ignore_prices_with_zero_volumes: True
ignore_zero_prices: True
ignore_negative_prices: False
#
# Price frequency (we collect daily data, and separately this frequency
intraday_frequency: H
#
dont_sample_daily_if_hourly_fails: True
#

[robcarver17]

Moved into function but not yet configurable and implements old behaviour

def apply_price_cleaning(data: dataBlob,
                          broker_prices_raw: futuresContractPrices,
                          cleaning_control_args_override = arg_not_supplied):
    if broker_prices_raw is failure:
        return failure

    # FIXME NEEDS TO BE CONFIGURABLE
    ## It's important that the data is in local time zone so that this works
    broker_prices_no_future = broker_prices_raw.remove_future_data()

    ## Ignore zeros if no volumes (if volume could be real price eg crude oil)
    broker_prices = broker_prices_no_future.remove_zero_prices_if_zero_volumes()
    # END FIXME

    return broker_prices

priceFilterConfig = namedtuple('priceFilterConfig',
                               ['ignore_future_prices',
                                'ignore_prices_with_zero_volumes',
                                'ignore_zero_prices',
                                'ignore_negative_prices'])

def _get_config_for_price_filtering(data: dataBlob):
    production_config = get_production_config()
    ignore_future_prices = production_config.get_element_or_missing_data('ignore_future_prices')
    ignore_prices_with_zero_volumes = production_config.get_element_or_missing_data('ignore_future_prices')
    ignore_zero_prices = production_config.get_element_or_missing_data('ignore_zero_prices')
    ignore_negative_prices = production_config.get_element_or_missing_data('ignore_negative_prices')

    any_missing = any([x is arg_not_supplied for x in [ignore_future_prices, ignore_prices_with_zero_volumes, ignore_zero_prices, ignore_negative_prices]])

    if any_missing:
        error = 'Missing config items for price filtering - have you deleted from defaults.yaml?'
        data.log.critical(error)
        raise Exception(error)

    return priceFilterConfig(ignore_zero_prices=ignore_zero_prices,
                             ignore_negative_prices=ignore_negative_prices,
                             ignore_future_prices=ignore_future_prices,
                             ignore_prices_with_zero_volumes=ignore_prices_with_zero_volumes)

[robcarver17]

def apply_price_cleaning(data: dataBlob,
                          broker_prices_raw: futuresContractPrices,
                          cleaning_control_args_override = arg_not_supplied):

    cleaning_config = _get_config_for_price_filtering(data =data,
                                                      cleaning_control_args_override=cleaning_control_args_override)

    if broker_prices_raw is failure:
        return failure

    broker_prices = copy(broker_prices_raw)

    ## It's important that the data is in local time zone so that this works
    if cleaning_config.ignore_future_prices:
        broker_prices = broker_prices.remove_future_data()

    if cleaning_config.ignore_prices_with_zero_volumes:
        broker_prices = broker_prices.remove_zero_volumes()

    if cleaning_config.ignore_zero_prices:
        ## need to implement
        broker_prices = broker_prices.remove_zero_prices()

    if cleaning_config.ignore_negative_prices:
        ## need to implement
        broker_prices = broker_prices.remove_negative_prices()

    return broker_prices

That's the final filtering applied.

[robcarver17]

We can now change the configuration for the cleaning when we run interactively. Will also make debugging easier.

Let's wrap the spike check and the hourly behaviour into the same thing.

[robcarver17]

def update_historical_prices_for_instrument_and_contract(
    contract_object: futuresContract, data: dataBlob,
        cleaning_config: priceFilterConfig = arg_not_supplied
):

    diag_prices = diagPrices(data)
    intraday_frequency = diag_prices.get_intraday_frequency_for_historical_download()
    daily_frequency = DAILY_PRICE_FREQ

    # Get *intraday* data (defaults to hourly)
    result = get_and_add_prices_for_frequency(
        data,
        contract_object,
        frequency=intraday_frequency,
        cleaning_config=cleaning_config
    )
    if result is failure:
        # Skip daily data if intraday not working
        if cleaning_config.dont_sample_daily_if_intraday_fails: ########### <- now configurable behavour
            return failure

    # Get daily data
    # we don't care about the result flag for this
    get_and_add_prices_for_frequency(data, contract_object,
                                     frequency=daily_frequency,
                                     cleaning_config=cleaning_config)

    return success

def get_and_add_prices_for_frequency(
    data: dataBlob,
    contract_object: futuresContract,
    frequency: Frequency = DAILY_PRICE_FREQ,
    cleaning_config: priceFilterConfig = arg_not_supplied
):
    broker_data_source = dataBroker(data)
    db_futures_prices = updatePrices(data)

    broker_prices = broker_data_source.get_cleaned_prices_at_frequency_for_contract_object(
        contract_object, frequency
    )

    # FIXME NEED FAILURE MODE INSTEAD
    if len(broker_prices) == 0:
        data.log.msg("No prices from broker for %s" % str(contract_object))
        return failure

    error_or_rows_added = db_futures_prices.update_prices_for_contract(
        contract_object, broker_prices, check_for_spike=True,
        max_price_spike = cleaning_config.max_price_spike ########### <- max price spike passed down
    )

    if error_or_rows_added is spike_in_data:
        report_price_spike(data, contract_object)
        return failure

    data.log.msg(
        "Added %d rows at frequency %s for %s"
        % (error_or_rows_added, frequency, str(contract_object))
    )
    return success

[robcarver17]

Think that's the code written, now to test.

[robcarver17]

Option to override configuration is now optional

[robcarver17]

Testing hasn't thrown up any problems.
Added a fair amount of logging.

Next step: document

[robcarver17]

Documentation done.

[robcarver17]

Error 162, reqId 15: Historical Market Data Service error message:HMDS query returned no data: CZKH3@GLOBEX Trades, contract: Contract(secType='FUT', conId=513880723, symbol='CZK', lastTradeDateOrContractMonth='20230313', multiplier='4000000', exchange='GLOBEX', currency='USD', localSymbol='CZKH3', tradingClass='CZK')
2022-06-15 08:36:06 {'type': 'Update-Historical-prices-manually', 'instrument_code': 'CZK', 'contract_date': '20220900', 'broker': 'IB', 'clientid': 1172, 'component': 'ibFuturesContractPriceData'}  Reqid 15: 162 Historical Market Data Service error message:HMDS query returned no data: CZKH3@GLOBEX Trades  (CZK/20230313)
2022-06-15 08:36:08 {'type': 'Update-Historical-prices-manually', 'instrument_code': 'CZK', 'contract_date': '20230300', 'broker': 'IB', 'clientid': 1172, 'component': 'ibFuturesContractPriceData'} [Warning] No price data from IB
2022-06-15 08:36:10 {'type': 'Update-Historical-prices-manually', 'instrument_code': 'CZK', 'contract_date': '20230300', 'broker': 'IB', 'clientid': 1172, 'component': 'ibFuturesContractPriceData'} [Warning] Something went wrong getting IB price data for CZK/20230300

These error 162's are being treated as a problem rather than zero volume. As a result a lot of instruments are tripping from hourly without collecting daily. I suppose if there is no hourly volume there is no daily volume?