//+------------------------------------------------------------------+
//|                                                     DataHarvester.mqh |
//|                 Centaur Quant Architecture — Data Module           |
//|                Lightweight Live Market Telemetry Harvester         |
//+------------------------------------------------------------------+
//|  PURPOSE                                                          |
//|  Captures Bid/Ask/Spread/TickVolume on a controlled cadence and   |
//|  packages them as SDP Tick_Harvest frames via CSDPEncoder. Emits  |
//|  periodic SDP Heartbeat frames. Zero blocking, O(1) per pass —    |
//|  safe to call from OnTick/OnTimer. Dispatch is decoupled: the     |
//|  caller drains pending frames via TakePending() and forwards them |
//|  through the transport (CSocketClient).                           |
//+------------------------------------------------------------------+
#property strict

#ifndef DATAHARVESTER_MQH
#define DATAHARVESTER_MQH

#include "../Network/CSDPEncoder.mqh"

//+------------------------------------------------------------------+
//|                                                                  |
//+------------------------------------------------------------------+
class CDataHarvester
  {
private:
   string            m_symbol;              // instrument being harvested
   string            m_timeframe;           // SDP wire timeframe (e.g., "M15")
   CSDPEncoder       m_encoder;             // SDP serializer — single source of truth

   ulong             m_harvest_interval_ms;   // min gap between Tick_Harvest encodes
   ulong             m_heartbeat_interval_ms; // period of Heartbeat emits
   ulong             m_last_harvest_ms;       // GetTickCount64() of last harvest
   ulong             m_last_heartbeat_ms;     // GetTickCount64() of last heartbeat
   bool              m_stale_logged;          // one-shot stale-tick warning flag

   bool              m_ready;                 // symbol validated

   //--- pending outbound queue (decoupled from transport) ---
   string            m_queue[];
   int               m_queue_head;
   enum { MAX_PENDING = 64, COMPACT_THRESHOLD = 512 };

   //--- telemetry counters ---
   ulong             m_ticks_harvested;
   ulong             m_heartbeats_emitted;

   string            PeriodName(const ENUM_TIMEFRAMES period);
   void              Enqueue(const string payload);
   void              CompactQueue();

public:
                     CDataHarvester(const string symbol, const string timeframe = "");
                    ~CDataHarvester();

   //--- one lightweight pass; call from OnTick or OnTimer ---
   bool              Harvest();

   //--- drain pending SDP payloads (caller forwards via CSocketClient) ---
   bool              HasPending() const;
   bool              TakePending(string &out_payload);
   int               PendingCount() const;

   //--- runtime configuration (clamped to sane minimums) ---
   void              SetHarvestInterval(const ulong interval_ms);
   void              SetHeartbeatInterval(const ulong interval_ms);

   //--- read access ---
   bool              IsReady() const { return m_ready; }
   string            Symbol() const { return m_symbol; }
   string            Timeframe() const { return m_timeframe; }
   ulong             TicksHarvested() const { return m_ticks_harvested; }
   ulong             HeartbeatsEmitted() const { return m_heartbeats_emitted; }
  };
//+------------------------------------------------------------------+
//| Constructor — bind symbol/timeframe, validate, reset timers.      |
//+------------------------------------------------------------------+
CDataHarvester::CDataHarvester(const string symbol, const string timeframe)
   : m_symbol(symbol),
     m_timeframe(timeframe),
     m_harvest_interval_ms(1000),
     m_heartbeat_interval_ms(30000),
     m_last_harvest_ms(0),
     m_last_heartbeat_ms(0),
     m_stale_logged(false),
     m_ready(false),
     m_queue_head(0),
     m_ticks_harvested(0),
     m_heartbeats_emitted(0)
  {
   if(StringLen(m_symbol) == 0)
     {
      PrintFormat("[CDataHarvester] ERROR: empty symbol in constructor.");
      return;
     }
   if(StringLen(m_timeframe) == 0)
      m_timeframe = PeriodName((ENUM_TIMEFRAMES)_Period);
// light validation: a live instrument exposes a positive point value
   m_ready = (SymbolInfoDouble(m_symbol, SYMBOL_POINT) > 0.0);
   if(!m_ready)
      PrintFormat("[CDataHarvester] ERROR: symbol '%s' not available in the terminal.", m_symbol);
   else
      PrintFormat("[CDataHarvester] INFO: ready on %s %s (harvest %I64u ms, heartbeat %I64u ms).",
                  m_symbol, m_timeframe, m_harvest_interval_ms, m_heartbeat_interval_ms);
  }
//+------------------------------------------------------------------+
//| Destructor — nothing to release; present for symmetry.            |
//+------------------------------------------------------------------+
CDataHarvester::~CDataHarvester()
  {
  }
//+------------------------------------------------------------------+
//| Harvest — single non-blocking pass.                               |
//| 1) Emits a Heartbeat frame when the heartbeat interval elapsed.   |
//| 2) Emits a Tick_Harvest frame when the harvest interval elapsed   |
//|    AND a fresh tick is available (stale ticks are skipped).       |
//| Returns true when at least one frame was queued.                  |
//+------------------------------------------------------------------+
bool CDataHarvester::Harvest()
  {
   if(!m_ready)
      return false;
   const ulong now = GetTickCount64();
   bool emitted = false;
//--- 1) periodic heartbeat ---
   if(m_last_heartbeat_ms == 0 || (now - m_last_heartbeat_ms) >= m_heartbeat_interval_ms)
     {
      Enqueue(m_encoder.EncodeHeartbeat(m_symbol, m_timeframe));
      m_last_heartbeat_ms = now;
      m_heartbeats_emitted++;
      emitted = true;
     }
//--- 2) throttled tick harvest ---
   if(m_last_harvest_ms == 0 || (now - m_last_harvest_ms) >= m_harvest_interval_ms)
     {
      MqlTick tick;
      if(!SymbolInfoTick(m_symbol, tick) || tick.bid <= 0.0 || tick.ask <= 0.0)
        {
         if(!m_stale_logged)
           {
            PrintFormat("[CDataHarvester] WARNING: no valid tick for %s. Harvest skipped.", m_symbol);
            m_stale_logged = true;
           }
         return emitted;
        }
      // fresh-tick gate: skip when the market is closed (stale last tick)
      if(tick.time <= 0 || (TimeCurrent() - tick.time) > 60)
        {
         if(!m_stale_logged)
           {
            PrintFormat("[CDataHarvester] WARNING: market closed or tick stale for %s. Harvest skipped.", m_symbol);
            m_stale_logged = true;
           }
         return emitted;
        }
      m_stale_logged = false;
      const string payload = m_encoder.EncodeTickHarvest(m_symbol, m_timeframe,
                             tick.bid, tick.ask,
                             tick.time, (long)tick.volume);
      if(StringLen(payload) > 0)
        {
         Enqueue(payload);
         m_ticks_harvested++;
         emitted = true;
        }
      m_last_harvest_ms = now;
     }
   return emitted;
  }
//+------------------------------------------------------------------+
//| Enqueue — append a payload to the pending queue.                  |
//| When full, the OLDEST pending frame is dropped (ring semantics).  |
//+------------------------------------------------------------------+
void CDataHarvester::Enqueue(const string payload)
  {
   if(StringLen(payload) == 0)
      return;
   const int size = ArraySize(m_queue);
   if(size - m_queue_head >= MAX_PENDING)
     {
      PrintFormat("[CDataHarvester] WARNING: pending queue full (%d). Oldest payload dropped.",
                  MAX_PENDING);
      m_queue_head++;
      return;
     }
   ArrayResize(m_queue, size + 1);
   m_queue[size] = payload;
  }
//+------------------------------------------------------------------+
//| HasPending — true when at least one frame awaits dispatch.        |
//+------------------------------------------------------------------+
bool CDataHarvester::HasPending() const
  {
   return (ArraySize(m_queue) - m_queue_head) > 0;
  }
//+------------------------------------------------------------------+
//| TakePending — pop the oldest pending frame; false when empty.     |
//| Compacted lazily to bound memory.                                 |
//+------------------------------------------------------------------+
bool CDataHarvester::TakePending(string &out_payload)
  {
   const int size = ArraySize(m_queue);
   if(m_queue_head >= size)
     {
      m_queue_head = 0;
      ArrayResize(m_queue, 0);
      return false;
     }
   out_payload = m_queue[m_queue_head++];
   if(m_queue_head >= COMPACT_THRESHOLD)
      CompactQueue();
   return true;
  }
//+------------------------------------------------------------------+
//| PendingCount — frames awaiting dispatch.                          |
//+------------------------------------------------------------------+
int CDataHarvester::PendingCount() const
  {
   return ArraySize(m_queue) - m_queue_head;
  }
//+------------------------------------------------------------------+
//| CompactQueue — shift live frames to the front and shrink.         |
//+------------------------------------------------------------------+
void CDataHarvester::CompactQueue()
  {
   const int size = ArraySize(m_queue);
   if(m_queue_head <= 0)
      return;
   const int remaining = size - m_queue_head;
   string tmp[];
   if(remaining > 0)
      ArrayCopy(tmp, m_queue, 0, m_queue_head, remaining);
   ArrayResize(m_queue, remaining);
   if(remaining > 0)
      ArrayCopy(m_queue, tmp);
   m_queue_head = 0;
  }
//+------------------------------------------------------------------+
//| SetHarvestInterval — throttle for Tick_Harvest frames.            |
//+------------------------------------------------------------------+
void CDataHarvester::SetHarvestInterval(const ulong interval_ms)
  {
   m_harvest_interval_ms = MathMax(interval_ms, (ulong)100);
  }
//+------------------------------------------------------------------+
//| SetHeartbeatInterval — period for Heartbeat frames.               |
//+------------------------------------------------------------------+
void CDataHarvester::SetHeartbeatInterval(const ulong interval_ms)
  {
   m_heartbeat_interval_ms = MathMax(interval_ms, (ulong)1000);
  }
//+------------------------------------------------------------------+
//| PeriodName — wire name of an ENUM_TIMEFRAMES value.               |
//+------------------------------------------------------------------+
string CDataHarvester::PeriodName(const ENUM_TIMEFRAMES period)
  {
   ENUM_TIMEFRAMES p = period;
   if(p == PERIOD_CURRENT)
      p = (ENUM_TIMEFRAMES)_Period;
   string name = EnumToString(p);       // e.g. "PERIOD_M15"
   StringReplace(name, "PERIOD_", "");  //      "M15"
   return name;
  }

#endif // DATAHARVESTER_MQH