Борьба за сохранение объектов в базе данных Django с помощью Celery Beat на Digital Ocean - PullRequest
Новая
       33

Борьба за сохранение объектов в базе данных Django с помощью Celery Beat на Digital Ocean

0 голосов
/ 10 октября 2018

Я пытаюсь сохранить объекты с помощью Celery Beat в моем приложении Django (показывая данные OHLC).

Этот скрипт отлично работает в локальной среде (сохраняет объекты 3M), но не в VPN, как Digital Ocean.Он сохраняет определенное количество объектов (примерно 200 тыс. Объектов или 2 ГБ), но затем удаляет другие объекты, чтобы добавить каждый новый объект, что полностью сбивает с толку.

Мой стек

  • Django
  • Redis
  • Супервизор
  • Ubuntu

Я не пользуюсь Supervisor в моем регионе, поэтому я думаю, что это вызывает проблему, но можетт опознать.Буду признателен за любые отзывы и помощь.

Сценарий 

@periodic_task(
    # run_every=(crontab(minute='*/1')),
    run_every=(crontab(minute='*/60')),
    name="load_data",
    ignore_result=False
)
def load_data():
# Forex OHLC
TOKEN = MYTOKEN
con = fxcmpy.fxcmpy(access_token = TOKEN, log_level = 'error')
start = dt.datetime(2010, 1, 1)
stop = dt.datetime.today()
df = pd.DataFrame(list(DatasourceItem.objects.filter(datasource__sub_category__exact='Forex').values('symbol')))

for i in df['symbol']:

    datasource_item_obj = DatasourceItem.objects.get(symbol=i)

    Ohlc.objects.filter(datasource = datasource_item_obj).delete()

    if datasource_item_obj.base_symbol:
        base_symbol = datasource_item_obj.base_symbol
        tar_symbol = datasource_item_obj.tar_symbol
        mod_symbol = base_symbol + "/" + tar_symbol
        sys_symbol = base_symbol + tar_symbol
    else:
        sys_symbol = datasource_item_obj.symbol
        mod_symbol = datasource_item_obj.symbol

    data = con.get_candles(mod_symbol, period='D1', start=start, stop=stop)
    del data['askopen']
    del data['askclose']
    del data['askhigh']
    del data['asklow']
    del data['tickqty']
    data.columns = ['Open', 'Close', 'High', 'Low']
    data = data[['Open', 'High', 'Low',  'Close']]
    data.insert(loc=0, column='Symbol', value=sys_symbol)
    data.reset_index(level=0, inplace=True)
    data.dropna()
    # .values = return numpy array
    data_list = data.values.tolist()
    for row in data_list:
        new_price = Ohlc(time = row[0], symbol = row[1], open_price = row[2], high_price = row[3], low_price = row[4], close_price = row[5], datasource = datasource_item_obj)
        new_price.save()

# Stock OHLC
start = dt.datetime.now() - dt.timedelta(days=(365.25*5))
stop = dt.datetime.today()

df = pd.DataFrame(list(DatasourceItem.objects.filter(datasource__sub_category__exact='Stock').values('symbol')))
for i in df['symbol']:
    datasource_obj = DatasourceItem.objects.get(symbol=i)
    old_price = Ohlc.objects.filter(datasource = datasource_obj).delete()

    symbol = datasource_obj.symbol
    data = get_historical_data(symbol, start=start, stop=stop, output_format='pandas')
    del data['volume']
    data.columns = ['Open', 'High', 'Low', 'Close']
    data.insert(loc=0, column='Symbol', value=symbol)
    data.reset_index(level=0, inplace=True)
    data.dropna()
    data_list = data.values.tolist()
    for row in data_list:
        price = Ohlc(time = row[0], symbol = row[1], open_price = row[2], high_price = row[3], low_price = row[4], close_price = row[5], datasource = datasource_obj)
        price.save()

1 Ответ

0 голосов
/ 10 октября 2018

Эй, это происходит из-за количества транзакций, происходящих с базой данных, поэтому попробуйте оптимизировать запрос на создание данных, например, вы можете использовать массовое создание вместо того, чтобы создавать каждый объект отдельно.

price_list
for row in data_list:
    price = Ohlc(time = row[0], symbol = row[1], open_price = row[2], high_price = row[3], low_price = row[4], close_price = row[5], datasource = datasource_obj)
    price_list.append(price)
Ohlc.objects.bulk(price_list)

возможно, он не будет создавать большие данные за один раз, а затем разбивать данные на порции 1000.

...