TSAStreams.h

/*  ===================================================================
*
*                 T R A D I N G   S Y S T E M   A P I ™
*    Copyright © 1999 - 2014 by Peter Ritter ( TradingSystemAPI.com )
*                A L L   R I G H T S   R E S E R V E D
*
*     Consult your license regarding permissions and restrictions.
*             You may obtain a copy of the License at:
*     http://www.TradingSystemAPI.com/licenses/LICENSE-TSAPI-3.0.html
*
*  ====================================================================
*/

#ifndef TSA_RECORDSTREAMS__INCLUDED
#define TSA_RECORDSTREAMS__INCLUDED

#include <vector>

#include "TSASeriesTemplate.h"
#include "TSASeriesProxy.h"
#include "TSAComponent.h"
#include "TSADataDef.h"


namespace tsa {

    class in_stream;
    class instrument;
    class instrument_price_bar;
    class record;

    namespace fast {
        class database;
    }

    /*
    ** ====================================================
    ** >>>          class in_stream_adaptor            <<<
    ** ====================================================
    */

    class dll_export in_stream_adaptor
        : public component {
        friend class strategy;
        friend class instrument;
        friend class in_stream;
      private:
        bool m_bOpenExists = false;
        size_t m_nOpenIdx;
        bool m_bHighExists = false;
        size_t m_nHighIdx;
        bool m_bLowExists = false;
        size_t m_nLowIdx;
        bool m_bCloseExists = false;
        size_t m_nCloseIdx;
        bool m_bVolumeExists = false;
        size_t m_nVolumeIdx;
        bool m_bOpenInterestExists = false;
        size_t m_nOpenInterestIdx;
        bool m_bBidExists = false;
        size_t m_nBidIdx;
        bool m_bAskExists = false;
        size_t m_nAskIdx;
        bool m_bBidSizeExists = false;
        size_t m_nBidSizeIdx;
        bool m_bAskSizeExists = false;
        size_t m_nAskSizeIdx;
        bool m_bLastExists = false;
        size_t m_nLastIdx;
        bool m_bPriceExists = false;
        size_t m_nPriceIdx;
        bool m_bValueExists = false;
        size_t m_nValueIdx;
        variant m_CurrentValue;
        std::string m_stream_name;
        bool m_bIsConnected = false;
        bool m_bIsStrategyScheduler = false;
      public:
        //bool m_field_are_analyzed = false;
        enum price_bar_fields_availability {
            not_analyzed,
            ohlc_fields_exist,
            hlc_fields_exist,
            close_field_exists,
            last_field_exists
        };
        price_bar_fields_availability m_price_bar_field_availability = not_analyzed;
        bool determine_price_bar_field_availability(void);
        void update_price_bar(instrument_price_bar& price_bar);

      public:
        in_stream_adaptor(void);

        ~in_stream_adaptor(void);

        void stream_name(const std::string& stream_name);

        const std::string& stream_name(void) const;

        void current_value(const variant& value);
      public:

        void set_connected(bool is_connected_flag = true);

        bool is_connected(void) const;

        void InitFieldIndices(void);
      public:

        size_t open_field_position(void) const;

        size_t high_field_position(void) const;

        size_t low_field_position(void) const;

        size_t close_field_position(void) const;
      public:
        //to be called instead of OnOpen()
        void on_open__prepare_auto_series(void);
        virtual bool isa__get_ohlc_fields_exist(void);    //does not need to be overridden
        //=====================================
        //=====================================

        virtual void isa__set_forward_shift(size_t);

        virtual void isa__set_position(const date_time&) = 0;

        virtual bool isa__field_exists(const std::string&) = 0;

        virtual type_t isa__field_type(size_t) = 0;

        virtual size_t isa__field_index(const std::string&) = 0;

        virtual size_t isa__field_count() = 0;

        virtual const std::string& isa__field_name(size_t) = 0;

        virtual date_time isa__get_record_timestamp(void) = 0;

        virtual date_time isa__begin(void) = 0;

        virtual date_time isa__end(void) = 0;

        virtual size_t isa__size(void) = 0;

        virtual void isa__on_open(void) = 0;

        virtual variant isa__get_variant(size_t idx) = 0;

        virtual double isa__get_double(size_t idx) = 0;

        virtual void isa__update_value_at_position(size_t idx) = 0;

      public:

        virtual void isa__get_price_bar(const date_time& date_time, instrument_price_bar& priceBar) = 0;

        virtual bool isa__get_time_of_next_bar(date_time& nextDateTime,
                                               const date_time& currentDateTime) = 0;

        virtual std::string isa__stream_name(void) const = 0;

        virtual bool isa__implements_scheduling(void) const = 0;

        //=================
        //=================
      public:

        bool get_is_strategy_scheduler(void) const {
            return m_bIsStrategyScheduler;
        };
        void set_is_strategy_scheduler(bool _b) {
            m_bIsStrategyScheduler = _b;
        }
      protected:
        size_t m_nOpenIdx_PMS, m_nHighIdx_PMS, m_nLowIdx_PMS, m_nCloseIdx_PMS;
      public:
        virtual void cmp__finalize(void) override {
            ;
        }
        virtual void cmp__evaluate_bar(const date_time&) override;
        virtual void cmp__master_evaluate_bar(const date_time&) override;
      private:
        void raise_field_error(const std::string&) const;
      public:
        std::vector<series_core> m_auto_series;
      public:
        const series_core& open(void) const ;
        const series_core& high(void) const;
        const series_core& low(void) const;
        const series_core& close(void) const;
        const series_core& volume(void) const;
        const series_core& bid(void) const;
        const series_core& ask(void) const;
        const series_core& bidsize(void) const;
        const series_core& asksize(void) const;
        const series_core& last(void) const;
        const series_core& price(void) const;
        const series_core& value(void) const;
        const series_core& open_interest(void) const;
        const series_core& at(size_t) const;
    };



    /*
    ** ====================================================
    ** >>>             class in_stream                  <<<
    ** ====================================================
    */

    class dll_export in_stream : public object {
        tsa_declare_testable;
      protected:
        //bool m_is_init;
        bool m_owns_pointer;
        in_stream_adaptor* m_stream_adaptor_ptr = nullptr;
      public:
        in_stream(void);

        in_stream(const in_stream&);

        virtual ~in_stream(void);

        in_stream_adaptor* adaptor_ptr(void) const;

        in_stream_adaptor& adaptor(void) const;

        //IU
        bool Connected(void) const;

        bool is_init(void) const;

        void verify_is_init(void) const;

        const std::string& source(void) const;
        void as_random_walk(duration d = duration::days(1));

        void connect(const std::string& stream_name);
        void connect(const std::string& stream_name, const std::string& data_store_name);

        void set_forward_shift(size_t shift_);

        void init_auto_series(void);
      public:
        bool field_exists(const std::string& field_name) const;

        size_t field_index(const std::string& field_name) const;

        size_t field_count(void) const;

        const std::string& field_name(size_t index) const;

        tsa::type_t field_type(const std::string& field_name)const;
      public:

        date_time begin(void);

        date_time end(void);

        size_t size(void);

      public:
        void connect(in_stream_adaptor& stream_adaptor);
      public:
        //IU
        void connect(strategy*, fast::database&, const std::string& sourceName);
        //IU
        void set_default_stream_(void);
      private:
        // IU
        void as_time_master(void);
      public:

        const series_core& operator()(size_t index) const;

        const series_core& operator()(const std::string& field_name_) const;

        date_time timestamp(void) const;
      public:
        //======================
        //======================
        series<double> open;
        series<double> high;
        series<double> low;
        series<double> close;
        series<double> volume;
        series<double> bid;
        series<double> ask;
        series<double> bidsize;
        series<double> asksize;
        series<double> last;
        series<double> price;
        series<double> value;
        series<double> openint;
        //
        //auto_series close2;
    };


    enum rec_stream_open_flag {
        replace,      
        no_replace    
    };

    /*
    ** ====================================================
    ** >>>          class out_stream_adaptor           <<<
    ** ====================================================
    */

    class dll_export out_stream_adaptor : public component {
      public:

        virtual ~out_stream_adaptor(void) {
            ;
        }
      public:

        virtual void osa__append(const date_time& time_stamp,
                                 std::vector<variant>& record) = 0;

        virtual void osa__prepare_stream(const std::string& stream_name,
                                         const column_defs& field_list) = 0;// , rec_stream_open_flag open_flag) = 0;
      public:

        virtual void cmp__finalize(void) override {
            ;
        }
    };


    /*
    ** ====================================================
    ** >>>            class out_stream                  <<<
    ** ====================================================
    */

    class dll_export out_stream :  public component, public object {
      private:
        out_stream_adaptor* m_out_stream_adaptor_ptr;
        bool m_owns_adaptor_ptr;
        std::string m_stream_name;
        bool m_is_init;
      private:
        bool m_bFirstRecordCompletedAndSchemaReset;
        bool m_bPrevValueWasFieldName;
        size_t m_nInputPos;
        size_t m_nValuePos;
        size_t m_nCurrInputPos;
        size_t m_nNumNamedFields;
        size_t m_nNumValues;
        std::string m_sMostRecentFieldName;
        std::vector<type_t> m_Types;
        std::vector<bool> m_IsValue;
        std::vector<std::string> m_FieldName;
        std::vector<variant> m_RecordSink_Values;
        date_time m_RecordSink_Timestamp;
      public:
        out_stream(void);
      private:
        out_stream(const out_stream&) = delete;
      public:
        virtual ~out_stream(void);
      private:
        void operator= (out_stream_adaptor&);   //disabled
      public:
        void connect(const std::string& table_name);// , rec_stream_open_flag flag = no_replace);
        void connect(const std::string& table_name, const std::string& data_store_name);

        void attach(out_stream_adaptor& stream_adaptor);
      public:
        out_stream& operator<< (const char* char_string);

        out_stream& operator<< (const tsa::string& char_string);

        out_stream& operator<< (const std::string& char_string);
        out_stream& operator<< (int num);
        out_stream& operator<< (unsigned num);
        out_stream& operator<< (int64_t num);
        out_stream& operator<< (uint64_t num);
        out_stream& operator<< (float num);
        out_stream& operator<< (double num);
        out_stream& operator<< (bool flag);
        out_stream& operator<< (const date_time& date_time);
        out_stream& operator<< (const date& dt);
        out_stream& operator<< (const variant& var);//for NULL/undefined values.

        template<typename T>
        out_stream& operator<< (const series<T>& ser) {
            operator<<(ser[0]);
            return *this;
        }
      public:
        void ProcessFirstLineValue(const variant& var);
        void ProcessValue(const variant&);
      private:
        void ResetCounters(void);
        bool FirstRecordCompletedAndSchemaReset(void) const;
      public:
        virtual void cmp__post_evaluate_bar(const date_time&) override;
    };



    /*
    ** ====================================================
    ** >>>      class in_stream_adaptor_platform       <<<
    ** ====================================================
    */

    // Internal
    class dll_export in_stream_adaptor_platform
        : public in_stream_adaptor, public tsa::object {
        double prices[5];
        std::string m_sFieldName;
      public:
        date_time m_CurrentTimestamp;
      public:
        in_stream_adaptor_platform(void);
        virtual ~in_stream_adaptor_platform(void);
      public:
        void set_timestamp(const date_time&);
        void set_prices(double _o, double _h, double _l, double _c, double _v);
        void SetData(const date_time& _ts, double _o, double _h, double _l, double _c, double _v);
      public:
        virtual void isa__get_price_bar(const date_time& date_time,
                                        instrument_price_bar& priceBar) override;
        virtual bool isa__get_time_of_next_bar(date_time& nextDateTime,
                                               const date_time& currentDateTime) override;
        virtual std::string isa__stream_name(void) const override;
        virtual bool isa__implements_scheduling(void) const override;
        //virtual bool isa__ImplementsPriceMaster(void) const override;
      public:
        virtual void isa__set_position(const date_time&) override;
        virtual bool isa__field_exists(const std::string&) override;
        virtual type_t isa__field_type(size_t) override;
        virtual const std::string& isa__field_name(size_t) override;
        virtual size_t isa__field_index(const std::string&) override;
        virtual size_t isa__field_count(void) override;
        virtual date_time isa__get_record_timestamp(void) override;
        virtual date_time isa__begin(void) override;
        virtual date_time isa__end(void) override;
        virtual size_t isa__size(void) override;
        virtual void isa__on_open(void) override {
            ;
        }
      public:
        virtual variant isa__get_variant(size_t idx) override;
        virtual double isa__get_double(size_t idx) override;
        virtual void isa__update_value_at_position(size_t idx) override;
    };




} // namespace tsa
#endif