Setting up new system to use parquet, error I can't figure out

[t3chap]

I had previously setup a working system in a docker container. Since it seemed to be working, I'm setting up a production box that will automatically load mongodb as a service, cron jobs, etc, which might be possible in docker but are easier and more robust in ubuntu server by itself.
I copied the setup files from my previous working setup and the backtests work fine (and agree with the prior setup).
When I start to run production code

from sysproduction import update_system_backtests
update_system_backtests.update_system_backtests()

I get the following error, which I'm having trouble determining why:

Traceback (most recent call last):
  File "<stdin>", line 1, in <module>
  File "/home/tim/pysystemtrade/sysproduction/update_system_backtests.py", line 29, in update_system_backtests
    system_backtest_runner.run_strategy_method()
  File "/home/tim/pysystemtrade/syscontrol/strategy_tools.py", line 41, in run_strategy_method
    method()
  File "/home/tim/pysystemtrade/sysproduction/strategy_code/run_system_classic.py", line 64, in run_backtest
    function_to_call_on_update(
  File "/home/tim/pysystemtrade/sysproduction/strategy_code/run_system_classic.py", line 153, in updated_buffered_positions
    lower_buffer, upper_buffer = get_position_buffers_from_system(
                                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/sysproduction/strategy_code/run_system_classic.py", line 177, in get_position_buffers_from_system
    buffers = system.portfolio.get_buffers_for_position(
              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 148, in get_buffers_for_position
    position = self.get_notional_position(instrument_code)
               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 211, in get_notional_position
    self.get_notional_position_before_risk_scaling(instrument_code)
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 237, in get_notional_position_before_risk_scaling
    notional_position_without_idm = self.get_notional_position_without_idm(
                                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 252, in get_notional_position_without_idm
    instr_weights = self.get_instrument_weights()
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 433, in get_instrument_weights
    self.get_unsmoothed_instrument_weights_fitted_to_position_lengths()
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 453, in get_unsmoothed_instrument_weights_fitted_to_position_lengths
    raw_instrument_weights = self.get_unsmoothed_raw_instrument_weights()
                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 497, in get_unsmoothed_raw_instrument_weights
    raw_instrument_weights = self.get_raw_fixed_instrument_weights()
                             ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 539, in get_raw_fixed_instrument_weights
    instrument_weights_dict = self.get_fixed_instrument_weights_from_config()
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 562, in get_fixed_instrument_weights_from_config
    instrument_weights_dict = self._add_zero_instrument_weights(
                              ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 589, in _add_zero_instrument_weights
    self.allocate_zero_instrument_weights_to_these_instruments()
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 753, in allocate_zero_instrument_weights_to_these_instruments
    instruments_without_data_or_weights = self.instruments_without_data_or_weights()
                                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 814, in instruments_without_data_or_weights
    subsystem_positions = copy(self._get_all_subsystem_positions())
                               ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 616, in _get_all_subsystem_positions
    positions = self.get_subsystem_positions_for_instrument_list(instrument_list)
                ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 478, in get_subsystem_positions_for_instrument_list
    subsystem_positions = [
                          ^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 479, in <listcomp>
    self.get_subsystem_position(instrument_code)
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 703, in wrapper
    return func(*args, **kwargs)
           ^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/portfolio.py", line 857, in get_subsystem_position
    return self.position_size_stage.get_subsystem_position(instrument_code)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/positionsizing.py", line 125, in get_subsystem_position
    vol_scalar = self.get_average_position_at_subsystem_level(instrument_code)
                 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/positionsizing.py", line 200, in get_average_position_at_subsystem_level
    instr_value_vol = self.get_instrument_value_vol(instrument_code)
                      ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/positionsizing.py", line 240, in get_instrument_value_vol
    instr_ccy_vol = self.get_instrument_currency_vol(instrument_code)
                    ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/positionsizing.py", line 282, in get_instrument_currency_vol
    block_value = self.get_block_value(instrument_code)
                  ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 726, in wrapper
    ans = system.cache.calc_or_cache(
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/system_cache.py", line 573, in calc_or_cache
    value = func(this_stage, *args, **kwargs)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/home/tim/pysystemtrade/systems/positionsizing.py", line 321, in get_block_value
    value_of_price_move = self.rawdata_stage.get_value_of_block_price_move(
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
AttributeError: 'RawData' object has no attribute 'get_value_of_block_price_move'

It almost seems like something that should be in mongodb or parquet isn't getting populated, but I can't figure out what it is.
Anyone have any ideas?
I've verified my mongodb is running, and the directory is filling up with files. I'm able to manually read the parquet files.
I'm wondering if I didn't initialize something that was supposed to go into a database. I have verified that the csv files that have the block info are on the system.

[t3chap]

In looking at mongodb objects, in the prior system that didn't throw the error I have these mongodb items:

['spread_costs', '_STRATEGY_HISTORIC_ORDERS', '_CONTRACT_HISTORIC_ORDERS', 'limit_status', 'CONTRACT_ORDER_STACK', 'IBClientTracker', 'futures_roll_status', 'process_control', 'INSTRUMENT_ORDER_STACK']

In the one giving me the error I only have these items:

['spread_costs', 'IBClientTracker']

Perhaps this has something to do with my errors?
Or it could be they don't exist because the update_backtests failed so nothing created them?

UPDATE
I did a mongodump and restore and still am having the same error.

[t3chap]

My rawdata.py is one I borrowed from Rorymac, and when I added

    def get_raw_cost_data(self, instrument_code: str):
        return self.data_stage.get_raw_cost_data(instrument_code)

    def get_value_of_block_price_move(self, instrument_code: str) -> float:
        return self.data_stage.get_value_of_block_price_move(instrument_code)

    def get_fx_for_instrument(self, instrument_code: str, base_currency: str):
        return self.data_stage.get_fx_for_instrument(
            instrument_code=instrument_code, base_currency=base_currency
        )

I was able to get it working. I also reversed commit 1011459.
I'll have to look at this again, I'm about done for today. I'm still learning the structure and it took me a while to figure this one out.
Hopefully this will help someone else.

[t3chap]

The following below should work, with some minor tweaks if you have backups that are csv. I modified the pysystemtrade/sysinit/contract_prices_from_csv_to_arctic.py file to not require user consent. In the pysystemtrade/sysinit/ directory you will also find files that will move the multiple files and adjusted files if you have the csv versions backed up. While the files say arctic in them they will go to parquet with the new setup, they weren't all renamed, probably because they would break things.You are required to have your parquet directory correct in your private/private_config.yaml file, eg
parquet_store: '/home/tim/data/parquet/'

from syscore.constants import arg_not_supplied

from sysdata.csv.csv_futures_contract_prices import csvFuturesContractPriceData
from sysproduction.data.prices import diagPrices
from sysobjects.contracts import futuresContract

diag_prices = diagPrices()


def init_db_with_csv_futures_contract_prices(
    datapath: str, csv_config=arg_not_supplied
):
    csv_prices = csvFuturesContractPriceData(datapath)
    input(
        "WARNING THIS WILL ERASE ANY EXISTING DATABASE PRICES WITH DATA FROM %s ARE YOU SURE?! (CTRL-C TO STOP)"
        % csv_prices.datapath
    )

    instrument_codes = csv_prices.get_list_of_instrument_codes_with_merged_price_data()
    instrument_codes.sort()
    for instrument_code in instrument_codes:
        init_db_with_csv_futures_contract_prices_for_code(
            instrument_code, datapath, csv_config=csv_config
        )

def init_db_with_csv_futures_contract_prices_no_stop(
#Import csv files to db without asking for confirmation

    datapath: str, csv_config=arg_not_supplied,verbose=True
):
    csv_prices = csvFuturesContractPriceData(datapath)
    instrument_codes = csv_prices.get_list_of_instrument_codes_with_merged_price_data()
    instrument_codes.sort()
    for instrument_code in instrument_codes:
        init_db_with_csv_futures_contract_prices_for_code(
            instrument_code, datapath, csv_config=csv_config,
            verbose=verbose
        )



def init_db_with_csv_futures_contract_prices_for_code(
    instrument_code: str, datapath: str, csv_config=arg_not_supplied, verbose=True
):
    print(instrument_code)
    csv_prices = csvFuturesContractPriceData(datapath, config=csv_config)
    db_prices = diag_prices.db_futures_contract_price_data

    print("Getting .csv prices may take some time")
    csv_price_dict = csv_prices.get_merged_prices_for_instrument(instrument_code)

    print("Have .csv prices for the following contracts:")
    print(str(csv_price_dict.keys()))

    for contract_date_str, prices_for_contract in csv_price_dict.items():
        print("Processing %s" % contract_date_str)
        print(".csv prices are \n %s" % str(prices_for_contract))
        contract = futuresContract(instrument_code, contract_date_str)
        print("Contract object is %s" % str(contract))
        print("Writing to db")
        db_prices.write_merged_prices_for_contract_object(
            contract, prices_for_contract, ignore_duplication=True
        )
        print("Reading back prices from db to check")
        written_prices = db_prices.get_merged_prices_for_contract_object(contract)
        print("Read back prices are \n %s" % str(written_prices))


if __name__ == "__main__":
    input("Will overwrite existing prices are you sure?! CTL-C to abort")
    # modify flags as required
    datapath = "*** NEED TO DEFINE A DATAPATH***"
    init_db_with_csv_futures_contract_prices(datapath)

I've developed my own way to download csv from barchart so in the code below when the barchart files are referenced they are in fact csv files. Depending on what you have you might need some format changes.

you then need a file like this one to call for the previous one:

#'transferring files to parquet....'

SAVE_DIRECTORY = '/home/tim/data/barchart/barchart_daily/'

from sysdata.csv.csv_futures_contract_prices import ConfigCsvFuturesPrices
from sysinit.futures.contract_prices_from_csv_to_arctic import (
    init_db_with_csv_futures_contract_prices_no_stop,
)


barchart_csv_config = ConfigCsvFuturesPrices(input_date_index_name="DateTime",
    input_skiprows=0,
    input_skipfooter=0,
    input_date_format="%Y-%m-%dT%H:%M:%S",
    #input_date_format = "%Y-%m-%d",
    input_column_mapping=dict(OPEN="open", HIGH="high", LOW="low", FINAL="close", VOLUME="volume"
    ),
)


def transfer_barchart_prices_to_parquet(datapath):
    init_db_with_csv_futures_contract_prices_no_stop(
        datapath, csv_config=barchart_csv_config,
        verbose = False
    )


transfer_barchart_prices_to_parquet(SAVE_DIRECTORY)

[rorymac]

Tim
Thanks I have tried a few other things and I think my best bet is to do a clean install, and try and populate the data from the csv's and copy the the db over without arctic.
The problem seems to be the environment so a clean install may fix all of those problems.

[t3chap]

Python environments can be tricky. I certainly didn't anticipate the mongodb version problem with some CPUs and the mint latest version not working with the mongodb later versions.

Most things in this area take at least twice what I think they should. Folks are usually helpful though.
Hope you get it sorted out.

[oldlore]

So I wasn't able to get the arctic data into parquet when I created a new VM with Mint 21.3 and python 3.10 in a venv. However, I opened an old copy of my current system (Mint 20.3, Python 3.8 (no venv), and using arctic) and was able to get the arctic data into parquet. I haven't tested it out fully to see if everything works.

1. git fetch then git pull to update code
2. ran the arctic to parquet .py file but it crashed on the global capital since mine is apparently empty. I only have my strategy capital data. So I changed the code to skip global capital and it worked.

Of course I still need to get this working in the newer mint and python version but at least getting the data into parquet helps me avoid having to do that again (I hope). Going to try setting up the new versions w/o venv to see if that makes any difference. I just don't know enough about virtual environments to troubleshoot them easily.

[rorymac]

Tim
Got it working see here : #1290 (reply in thread)