<commit_before>
// This serves as an abstraction layer for Gekko:
// we store all candles as 1m candles in a 
// database per day. Using this method you can
// 
// - Store new candles in the database based 
//   on fetched trade data.
// - Getting candles out of a database.
// 
// Notes:
// 
//  - All candles are stored in daily format
//    per exchange per market per day (in UTC).
//  - it adds empty candles to fill gaps, it
//    does this to differentiate between times
//    with no volume and missing data.
//  - because of this it expects empty candles
//    in the files, else it assumes there is
//    missing data (and the db is most likely
//    discarded as corrupt).
//  - It trusts the trade data provider to send
//    enough data: if the tradeFetcher doesn't
//    keep up with the trades coming in, this
//    manager cannot tell that the data is 
//    corrupted. And thus assumes it is not.
//  - The 1m candles are an internal datastructure
//    (referred to as fake candles), when clients
//    request candle data we convert the 1m
//    candles on the fly to the desired X minute
//    based candles (referred to as real candles).
//    
// Known issues:
//  - The manager is unable to correctly process
//    trade fetches that span over more than 2 days
//    (after filtering out all the trade data already
//    stored on disk).
//  - The leftovers are not persisted to disk: Gekko
//    uses leftovers to keep track of trades that
//    did not span a full minute in a certain fetch.
//    This makes sure we also store fully aggregated
//    minutes between two fetches. However this state
//    is stored in memory and is lost on restart. This
//    also means that we potentially lose accuracy in
//    the rare case that *on restart* the first fetches
//    are since the next-to-be-stored-candle but maybe
//    not since its start.

var _ = require('lodash');
var moment = require('moment');
var utc = moment.utc;
var log = require('./log.js');
var fs = require('fs');
var nedb = require('nedb');
var async = require('async');

var util = require('./util');
var config = util.getConfig();

var equals = util.equals;

// even though we have leap seconds, every
// day has the same amount of minutes
var MINUTES_IN_DAY = 1439;

var Manager = function() {
  _.bindAll(this);

  // all minutes we fetched that are part
  // of the real candle
  this.realCandleContents = [];

  this.historyPath = config.history.directory || './history/';
  this.days = {};
  this.realCandleSize = false;

  this.currentDay;

  // make sure the historical folder exists
  if(!fs.existsSync(this.historyPath))
    fs.mkdirSync(this.historyPath);

  this
    .on('processed', function() {
      this.defer(function() {
        log.info(
          'Processed trades, sleeping for',
          this.fetch.next.m.fromNow(true),
          '...'
        );
      });
    })
    .on('fake candle', this.watchRealCandles)
    .on('real candle', function(candle) {
      // log.debug(
      //   'NEW REAL CANDLE:',
      //   [
      //     '',
      //     'TIME:\t' + candle.start.format('YYYY-MM-DD HH:mm:ss'),
      //     'O:\t' + candle.o,
      //     'H:\t' + candle.h,
      //     'L:\t' + candle.l,
      //     'C:\t' + candle.c,
      //     'V:\t' + candle.v,
      //     'VWP:\t' + candle.p
      //   ].join('\n\t')
      // );
    });
};

var Util = require('util');
var EventEmitter = require('events').EventEmitter;
Util.inherits(Manager, EventEmitter);

// load all databases we need based on fetch
// data (which tells us how for we can currently
// fetch back and thus what our limits are)
Manager.prototype.init = function(data) {
  // meta state on fetch data
  this.fetch = {
    start: this.mom(data.start),
    end: this.mom(data.end),
    next: this.mom(utc().add('ms', data.nextIn))
  };

  // what do we need if we want to start right 
  // away?
  this.required = {
    to: this.mom(utc().subtract('ms', data.timespan)),

    // todo: abstract EMA stuff out of this
    from: this.mom(util.intervalsAgo(config.EMA.candles).utc())
  }

  // the order in which we do things:
  // - (receive first batch)
  // - check historical db
  // - process new trades

  this.once('history', function() {
    this.processTrades(data);
  });

  this.setDay(this.fetch.start.m);
  this.loadDays();
}

Manager.prototype.setRealCandleSize = function(interval) {
  this.realCandleSize = interval;
}

// fires everytime we have a new 1m candle
// and aggregates them into the candles
// of the required size
Manager.prototype.watchRealCandles = function() {
  if(_.size(this.realCandleContents) < this.realCandleSize)
    return;

  var candle = this.calculateRealCandle(this.realCandleContents);

  this.emit('real candle', candle);

  this.realCandleContents = [];
}

// retrieve an 1m candle out of the DB
Manager.prototype.getCandle = function(mom, callback) {
  this.days[mom.dayString].handle.findOne({s: mom.minute}, callback);
}

Manager.prototype.setDay = function(m) {
  this.current = this.mom(m.clone());
}

Manager.prototype.today = function() {
  return utc().startOf('day');
}

// grab a batch of trades and for each full minute
// create a new candle
Manager.prototype.processTrades = function(data) {
  // first time make sure we have
  // loaded this day.
  if(!this.leftovers)
    this.loadDay(this.current.day);

  // if first time
  if(!this.minumum) {<commit_msg>fixed next fetch time calculation<commit_after>
// This serves as an abstraction layer for Gekko:
// we store all candles as 1m candles in a 
// database per day. Using this method you can
// 
// - Store new candles in the database based 
//   on fetched trade data.
// - Getting candles out of a database.
// 
// Notes:
// 
//  - All candles are stored in daily format
//    per exchange per market per day (in UTC).
//  - it adds empty candles to fill gaps, it
//    does this to differentiate between times
//    with no volume and missing data.
//  - because of this it expects empty candles
//    in the files, else it assumes there is
//    missing data (and the db is most likely
//    discarded as corrupt).
//  - It trusts the trade data provider to send
//    enough data: if the tradeFetcher doesn't
//    keep up with the trades coming in, this
//    manager cannot tell that the data is 
//    corrupted. And thus assumes it is not.
//  - The 1m candles are an internal datastructure
//    (referred to as fake candles), when clients
//    request candle data we convert the 1m
//    candles on the fly to the desired X minute
//    based candles (referred to as real candles).
//    
// Known issues:
//  - The manager is unable to correctly process
//    trade fetches that span over more than 2 days
//    (after filtering out all the trade data already
//    stored on disk).
//  - The leftovers are not persisted to disk: Gekko
//    uses leftovers to keep track of trades that
//    did not span a full minute in a certain fetch.
//    This makes sure we also store fully aggregated
//    minutes between two fetches. However this state
//    is stored in memory and is lost on restart. This
//    also means that we potentially lose accuracy in
//    the rare case that *on restart* the first fetches
//    are since the next-to-be-stored-candle but maybe
//    not since its start.

var _ = require('lodash');
var moment = require('moment');
var utc = moment.utc;
var log = require('./log.js');
var fs = require('fs');
var nedb = require('nedb');
var async = require('async');

var util = require('./util');
var config = util.getConfig();

var equals = util.equals;

// even though we have leap seconds, every
// day has the same amount of minutes
var MINUTES_IN_DAY = 1439;

var Manager = function() {
  _.bindAll(this);

  // all minutes we fetched that are part
  // of the real candle
  this.realCandleContents = [];

  this.historyPath = config.history.directory || './history/';
  this.days = {};
  this.realCandleSize = false;

  this.currentDay;

  // make sure the historical folder exists
  if(!fs.existsSync(this.historyPath))
    fs.mkdirSync(this.historyPath);

  this
    .on('processed', function() {
      this.defer(function() {
        log.info(
          'Processed trades, sleeping for',
          this.fetch.next.m.fromNow(true),
          '...'
        );
      });
    })
    .on('fake candle', this.watchRealCandles)
    .on('real candle', function(candle) {
      log.debug(
        'NEW REAL CANDLE:',
        [
          '',
          'TIME:\t' + candle.start.format('YYYY-MM-DD HH:mm:ss'),
          'O:\t' + candle.o,
          'H:\t' + candle.h,
          'L:\t' + candle.l,
          'C:\t' + candle.c,
          'V:\t' + candle.v,
          'VWP:\t' + candle.p
        ].join('\n\t')
      );
    });
};

var Util = require('util');
var EventEmitter = require('events').EventEmitter;
Util.inherits(Manager, EventEmitter);

// load all databases we need based on fetch
// data (which tells us how for we can currently
// fetch back and thus what our limits are)
Manager.prototype.init = function(data) {
  this.setFetchMeta(data);

  // what do we need if we want to start right 
  // away?
  this.required = {
    to: this.mom(utc().subtract('ms', data.timespan)),

    // todo: abstract EMA stuff out of this
    from: this.mom(util.intervalsAgo(config.EMA.candles).utc())
  }

  // the order in which we do things:
  // - (receive first batch)
  // - check historical db
  // - process new trades

  this.once('history', function() {
    this.processTrades(data);
  });

  this.setDay(this.fetch.start.m);
  this.loadDays();
}

Manager.prototype.setFetchMeta = function(data) {
  this.fetch = {
    start: this.mom(data.start),
    end: this.mom(data.end),
    next: this.mom(utc().add('ms', data.nextIn))
  };
}

Manager.prototype.setRealCandleSize = function(interval) {
  this.realCandleSize = interval;
}

// fires everytime we have a new 1m candle
// and aggregates them into the candles
// of the required size
Manager.prototype.watchRealCandles = function() {
  if(_.size(this.realCandleContents) < this.realCandleSize)
    return;

  var candle = this.calculateRealCandle(this.realCandleContents);

  this.emit('real candle', candle);

  this.realCandleContents = [];
}

// retrieve an 1m candle out of the DB
Manager.prototype.getCandle = function(mom, callback) {
  this.days[mom.dayString].handle.findOne({s: mom.minute}, callback);
}

Manager.prototype.setDay = function(m) {
  this.current = this.mom(m.clone());
}

Manager.prototype.today = function() {
  return utc().startOf('day');
}

// grab a batch of trades and for each full minute
// create a new candle
Manager.prototype.processTrades = function(data) {
  this.setFetchMeta(data);
  // first time make sure we have
  // loaded this day.
  if(!this.leftovers)
    this.loadDay(this.current.day);

  // if first time
  if(!this.minumum) {
{'fraction_non_alphanumeric': 0.12778288868445262, 'fraction_numerical': 0.0018399264029438822, 'mean_word_length': 2.6528897849462365, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7176112', 'n_tokens_mistral': 3468, 'n_tokens_neox': 3311, 'n_words': 1161}