Warrior_EA/System/AltDataFetch.mqh

//+------------------------------------------------------------------+
//|                                                 AltDataFetch.mqh |
//|                                                     AnimateDread |
//| missing history at attach time and keeps appending forward while |
//| deployed, so online learning never depends on an external Python |
//| process. research/altdata remains the RESEARCH side (screening,  |
//| new-source adjudication); this module maintains the exact same   |
//| files in production.                                             |
//+------------------------------------------------------------------+
#include "AltData.mqh"
#include "AtomicFile.mqh"

#define ALTFETCH_DIR            "Warrior_EA\\AltData\\"
#define ALTFETCH_RETRY_SECONDS  3600
#define ALTFETCH_GRID_START     D'2010.01.01'

//--- generic HTTP client (backoff/masking/JSON parse) and the symbol catalog + user-mapping
//--- persistence are separate collaborators - see those files' headers for why each is its own
//--- responsibility. Declared here, after ALTFETCH_DIR, since CAltDataCatalog's ctor path uses it.
#include "AltDataHttpClient.mqh"
#include "AltDataCatalog.mqh"

//--- one raw series: weekly COT positioning, a daily FRED series, or the EIA WPSR block
struct SAltRawSeries
  {
   datetime          date[];       // observation date, ascending
   datetime          published[];
   double            v1[];         // FRED: value | COT: open interest      | EIA: crude stocks ex-SPR
   double            v2[];         // COT: spec long                       | EIA: field production
   double            v3[];         // COT: spec short                      | EIA: refinery utilization %
   int               rows;
  };

//+------------------------------------------------------------------+
//| Append one empty row and return its index.                        |
//|                                                                  |
//| The five arrays are ONE record split across five buffers, so a    |
//| resize missed on any of them reads out of range on the next       |
//| append rather than failing here. Callers fill the values they     |
//| have; the rest stay 0.                                            |
//+------------------------------------------------------------------+
int AltSeriesAppend(SAltRawSeries &s, const datetime date, const datetime published)
  {
   int n = s.rows;
   ArrayResize(s.date, n + 1);
   ArrayResize(s.published, n + 1);
   ArrayResize(s.v1, n + 1);
   ArrayResize(s.v2, n + 1);
   ArrayResize(s.v3, n + 1);
   s.date[n] = date;
   s.published[n] = published;
   s.v1[n] = 0.0;
   s.v2[n] = 0.0;
   s.v3[n] = 0.0;
   s.rows = n + 1;
   return n;
  }

//--- Whole-file raw-series cache load/save (concern separated from the fetch pipelines that use
//--- it): pure functions, no member state - just the SAltRawSeries <-> ';'-delimited CSV mapping.

bool AltRawLoad(string name, SAltRawSeries &s, int cols)
  {
   s.rows = 0;
   int h = FileOpen(ALTFETCH_DIR + name, FILE_READ | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
   if(h == INVALID_HANDLE)
      return false;
   FileReadString(h); // header
   int cap = 2048;
   ArrayResize(s.date, cap);
   ArrayResize(s.published, cap);
   ArrayResize(s.v1, cap);
   ArrayResize(s.v2, cap);
   ArrayResize(s.v3, cap);
   while(!FileIsEnding(h))
     {
      string line = FileReadString(h);
      string f[];
      if(StringSplit(line, ';', f) < cols + 2)
         continue;
      if(s.rows >= cap)
        {
         cap = cap * 3 / 2;
         ArrayResize(s.date, cap);
         ArrayResize(s.published, cap);
         ArrayResize(s.v1, cap);
         ArrayResize(s.v2, cap);
         ArrayResize(s.v3, cap);
        }
      s.date[s.rows] = StringToTime(f[0]);
      s.published[s.rows] = StringToTime(f[1]);
      s.v1[s.rows] = StringToDouble(f[2]);
      s.v2[s.rows] = cols > 1 ? StringToDouble(f[3]) : 0.0;
      s.v3[s.rows] = cols > 2 ? StringToDouble(f[4]) : 0.0;
      s.rows++;
     }
   FileClose(h);
   return s.rows > 0;
  }

//--- Whole-file rewrites go through System\AtomicFile.mqh's temp + FileMove swap: multiple
//--- charts share these caches (raw_VIXCLS serves every symbol; two timeframes of one symbol
//--- share all of them), and a reader hitting a truncate-then-write mid-flight parses a torn file.
bool AltRawSave(string name, SAltRawSeries &s, int cols, string header)
  {
   string tmpName = "";
   int h = AtomicWriteBegin(ALTFETCH_DIR + name, FILE_COMMON, tmpName, FILE_TXT | FILE_ANSI);
   if(h == INVALID_HANDLE)
     {
      PrintFormat("AltDataFetch: cannot write %s (%d)", tmpName, GetLastError());
      return false;
     }
   FileWriteString(h, header + "\n");
   for(int i = 0; i < s.rows; i++)
     {
      string line = TimeToString(s.date[i], TIME_DATE) + ";" + TimeToString(s.published[i], TIME_DATE) + ";"
                    + DoubleToString(s.v1[i], 6);
      if(cols > 1)
         line += ";" + DoubleToString(s.v2[i], 6);
      if(cols > 2)
         line += ";" + DoubleToString(s.v3[i], 6);
      FileWriteString(h, line + "\n");
     }
   return AtomicWriteEnd(h, ALTFETCH_DIR + name, tmpName, FILE_COMMON, true, __FUNCTION__);
  }

class CAltDataFetch
  {
protected:
   string            m_fredKey;
   bool              m_keyLoaded;
   string            m_eiaKey;
   bool              m_eiaWarned;
   datetime          m_lastGexDate;   // last UTC day whose GEX row is already on disk
   bool              m_gexDateLoaded;
   datetime          m_lastAttempt[16]; // per-source retry throttle: 0=VIX 1=USD 2=COT 3=EIA 4=IVOL
                                        // 5=DGS10 6=T10Y2Y 7=T5YIE 8=DFF 9=ECBDFR 10=CPI 11=UNRATE

   //--- ---------------------------------------------------------- utils
   //--- Shared contract behind FredKey()/EiaKey(): the EA input is the key's home - it travels
   //--- with the EA and survives the Common\Files wipe that starts every fresh test (the
   //--- 2026-08-16 incident: keys.txt was wiped with the folder, FredKey() returned "" with NO
   //--- log line, COT updated but FRED never ran and the SP500 rebuild - gated on all three raws -
   //--- never fired). keys.txt is only the fallback for a blanked input, re-read on every
   //--- hourly-throttled attempt until a key is found; warnedFlag means "warned once", never fail
   //--- silently. fileMissingMsg == "" means "no distinct message for a missing file" (EIA's
   //--- shape) - the missing-file case then just falls through to notFoundMsg below.
   string            LoadCommonKey(const string inputValue, string &cache, bool &warnedFlag,
                                    const string prefix, const string fileMissingMsg, const string notFoundMsg)
     {
      if(inputValue != "")
         return inputValue;
      if(cache != "")
         return cache;
      int h = FileOpen(ALTFETCH_DIR + "keys.txt", FILE_READ | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
      if(h == INVALID_HANDLE)
        {
         if(fileMissingMsg != "")
           {
            if(!warnedFlag)
               Print(fileMissingMsg);
            warnedFlag = true;
            return "";
           }
        }
      else
        {
         while(!FileIsEnding(h))
           {
            string line = FileReadString(h);
            if(StringFind(line, prefix) == 0)
               cache = StringSubstr(line, StringLen(prefix));
           }
         FileClose(h);
        }
      if(cache == "" && !warnedFlag)
        {
         warnedFlag = true;
         Print(notFoundMsg);
        }
      return cache;
     }

   string            FredKey(void)
     {
      return LoadCommonKey(FredApiKey, m_fredKey, m_keyLoaded, "fred=",
                            "AltDataFetch: the FredApiKey input is blank and " + ALTFETCH_DIR + "keys.txt "
                            "does not exist in Common\\Files - FRED features (VIX, USD index) cannot update, "
                            "so the SP500/XAUUSD feature files will not build. Set the FredApiKey input "
                            "(free key: fred.stlouisfed.org) or create keys.txt with one line: fred=<key>. "
                            "The file is re-checked hourly; no restart needed.",
                            "AltDataFetch: the FredApiKey input is blank and " + ALTFETCH_DIR + "keys.txt has "
                            "no 'fred=<key>' line - FRED-based features will not update until one is provided "
                            "(free key: fred.stlouisfed.org).");
     }

   string            EiaKey(void)
     {
      return LoadCommonKey(EiaApiKey, m_eiaKey, m_eiaWarned, "eia=", "",
                            "AltDataFetch: the EiaApiKey input is blank and no 'eia=<key>' line exists in " +
                            ALTFETCH_DIR + "keys.txt - the EIA feature columns will stay empty (panel 0-fills; "
                            "free key: eia.gov/opendata). Re-checked hourly; no restart needed.");
     }

   //--- Shared gate at the top of every per-source Update method (Fred/Cot/Eia): skip while the
   //--- cached tail is still fresh (within staleDays), then skip if a fetch attempt for this
   //--- source's throttle slot already happened within ALTFETCH_RETRY_SECONDS - stamping the
   //--- attempt time only when neither check skips, so a genuine fetch below is exactly one
   //--- attempt per hour. UpdateGex has its own unrelated day/hour gate and does not use this.
   bool              ShouldAttemptFetch(const SAltRawSeries &s, int staleDays, int throttleSlot, datetime now)
     {
      if(s.rows > 0 && now - s.date[s.rows - 1] < staleDays * 86400)
         return false;
      if(m_lastAttempt[throttleSlot] != 0 && now - m_lastAttempt[throttleSlot] < ALTFETCH_RETRY_SECONDS)
         return false;
      m_lastAttempt[throttleSlot] = now;
      return true;
     }

   datetime          SaturdayAfter(datetime reportDate)
     {
      MqlDateTime st;
      TimeToStruct(reportDate, st);
      int days = (6 - st.day_of_week + 7) % 7;   // MQL5: 0=Sunday .. 6=Saturday
      if(days == 0)
         days = 7;
      return reportDate + days * 86400;
     }

   //--- ------------------------------------------------- FRED source appends observations after
   //--- the cached tail; full history when the cache is empty vmin/vmax: per-series plausibility
   //--- band. Rejected rows are counted and reported, never silently dropped.
   bool              UpdateFred(string seriesId, SAltRawSeries &s, int throttleSlot,
                                int staleDays = 4, double vmin = -1e18, double vmax = 1e18)
     {
      datetime now = TimeCurrent();
      //--- daily series: current until ~2 business days behind (weekends/holidays).
      //--- monthly series pass staleDays ~32 so a 30-day-old latest print does not
      //--- trigger a pointless fetch attempt every hour for weeks.
      if(!ShouldAttemptFetch(s, staleDays, throttleSlot, now))
         return false;
      string key = FredKey();
      if(key == "")
         return false;
      string url = "https://api.stlouisfed.org/fred/series/observations?series_id=" + seriesId +
                   "&api_key=" + key + "&file_type=json&limit=100000";
      //--- ALWAYS bound the start. 2005 leaves 5y of lookback margin ahead of the grid for the
      //--- longest window (3y COT percentile analog).
      string startDate = (s.rows > 0)
                         ? TimeToString(s.date[s.rows - 1] + 86400, TIME_DATE)
                         : "2005.01.01";
      StringReplace(startDate, ".", "-");      // TimeToString is yyyy.mm.dd; FRED wants ISO dashes
      url += "&observation_start=" + startDate;
      string body;
      if(!m_http.HttpGet(url, body))
         return false;
      int pos = 0, added = 0, rejected = 0;
      while(true)
        {
         int e1;
         string d = m_http.JsonField(body, "date", pos, e1);
         if(d == "" || e1 < 0)
            break;
         int e2;
         string v = m_http.JsonField(body, "value", e1, e2);
         pos = (e2 > e1 ? e2 : e1);
         if(v == "" || v == ".")
            continue;
         datetime dt = StringToTime(d);        // accepts yyyy-mm-dd
         if(dt <= 0 || (s.rows > 0 && dt <= s.date[s.rows - 1]))
            continue;
         double val = StringToDouble(v);
         if(val < vmin || val > vmax || !MathIsValidNumber(val))
           {
            rejected++;
            continue;
           }
         int n = AltSeriesAppend(s, dt, dt + 86400);   // unrevised daily: knowable next day
         s.v1[n] = val;
         added++;
        }
      if(rejected > 0)
         PrintFormat("AltDataFetch: %s REJECTED %d value(s) outside the plausibility band "
                     "[%g, %g] - transport corruption or format drift; investigate if repeated.",
                     seriesId, rejected, vmin, vmax);
      if(added > 0)
        {
         AltRawSave("raw_" + seriesId + ".csv", s, 1, "date;published;value");
         PrintFormat("AltDataFetch: %s +%d rows (through %s)", seriesId, added,
                     TimeToString(s.date[s.rows - 1], TIME_DATE));
        }
      else if(s.rows == 0)
         PrintFormat("AltDataFetch: %s returned a response with ZERO usable observations and "
                     "the cache is empty - series id wrong or API format changed. Features "
                     "depending on it will stay empty (panel 0-fills).", seriesId);
      return added > 0;
     }

   //--- -------------------------------------------------- COT source
   bool              UpdateCot(string cacheName, string cotWhere, SAltRawSeries &s, int throttleSlot)
     {
      datetime now = TimeCurrent();
      //--- weekly: a new Tuesday report exists once the last one is >7d old; allow 3 extra
      //--- days before nagging the API (Friday release + weekend)
      if(!ShouldAttemptFetch(s, 10, throttleSlot, now))
         return false;
      string where = cotWhere;
      if(s.rows > 0)
        {
         string lastDate = TimeToString(s.date[s.rows - 1], TIME_DATE);
         StringReplace(lastDate, ".", "-");    // ISO for the SoQL timestamp comparison
         where += " AND report_date_as_yyyy_mm_dd > '" + lastDate + "'";
        }
      string url = "https://publicreporting.cftc.gov/resource/gpe5-46if.json?%24select=" +
                   "report_date_as_yyyy_mm_dd,open_interest_all,lev_money_positions_long,lev_money_positions_short" +
                   "&%24where=" + m_http.UrlEncodePart(where) +
                   "&%24order=report_date_as_yyyy_mm_dd&%24limit=50000";
      string body;
      if(!m_http.HttpGet(url, body))
         return false;
      int pos = 0, added = 0;
      while(true)
        {
         int e1;
         string d = m_http.JsonField(body, "report_date_as_yyyy_mm_dd", pos, e1);
         if(d == "" || e1 < 0)
            break;
         int e2, e3, e4;
         string oi = m_http.JsonField(body, "open_interest_all", e1, e2);
         string ll = m_http.JsonField(body, "lev_money_positions_long", e2, e3);
         string ls = m_http.JsonField(body, "lev_money_positions_short", e3, e4);
         pos = (e4 > e1 ? e4 : e1);
         datetime dt = StringToTime(StringSubstr(d, 0, 10));
         if(dt <= 0 || oi == "" || ll == "" || ls == "")
            continue;
         if(s.rows > 0 && dt == s.date[s.rows - 1])
           {
            //--- same report date from a second contract-name variant: keep the larger-OI row
            if(StringToDouble(oi) > s.v1[s.rows - 1])
              {
               s.v1[s.rows - 1] = StringToDouble(oi);
               s.v2[s.rows - 1] = StringToDouble(ll);
               s.v3[s.rows - 1] = StringToDouble(ls);
              }
            continue;
           }
         if(s.rows > 0 && dt < s.date[s.rows - 1])
            continue;
         int n = AltSeriesAppend(s, dt, SaturdayAfter(dt));
         s.v1[n] = StringToDouble(oi);
         s.v2[n] = StringToDouble(ll);
         s.v3[n] = StringToDouble(ls);
         added++;
        }
      if(added > 0)
        {
         AltRawSave(cacheName, s, 3, "date;published;oi;lev_long;lev_short");
         PrintFormat("AltDataFetch: %s +%d weekly reports (through %s)", cacheName, added,
                     TimeToString(s.date[s.rows - 1], TIME_DATE));
        }
      else if(s.rows == 0)
         PrintFormat("AltDataFetch: COT query for %s returned ZERO reports on an empty cache - "
                     "the contract-name predicate probably matches nothing at the endpoint "
                     "(the like-predicates for non-SP500/JPY contracts are unverified; see the "
                     "catalog). COT features stay empty (panel 0-fills) until it is corrected.",
                     cacheName);
      return added > 0;
     }

   //--- ---------------------------------------------------- EIA source Weekly Petroleum Status
   //--- Report via the EIA v2 API, three series in one call (2010->now x 3 ~ 2,600 rows, under the
   //--- API's 5000-row page - revisit if series are added).
   bool              UpdateEia(SAltRawSeries &s, int throttleSlot)
     {
      datetime now = TimeCurrent();
      //--- weekly cadence, same nag guard as COT
      if(!ShouldAttemptFetch(s, 10, throttleSlot, now))
         return false;
      string key = EiaKey();
      if(key == "")
         return false;
      string startDate = TimeToString(s.rows > 0 ? s.date[s.rows - 1] + 86400 : ALTFETCH_GRID_START, TIME_DATE);
      StringReplace(startDate, ".", "-");      // TimeToString is yyyy.mm.dd; the API wants ISO dashes
      string url = "https://api.eia.gov/v2/petroleum/sum/sndw/data/?api_key=" + key +
                   "&frequency=weekly&data%5B0%5D=value" +
                   "&facets%5Bseries%5D%5B0%5D=WCESTUS1" +      // crude stocks ex-SPR
                   "&facets%5Bseries%5D%5B1%5D=WCRFPUS2" +      // field production
                   "&facets%5Bseries%5D%5B2%5D=WPULEUS3" +      // refinery utilization %
                   "&sort%5B0%5D%5Bcolumn%5D=period&sort%5B0%5D%5Bdirection%5D=asc" +
                   "&length=5000&start=" + startDate;
      string body;
      if(!m_http.HttpGet(url, body))
         return false;
      int pos = 0, added = 0;
      while(true)
        {
         int e1, e2, e3;
         string d = m_http.JsonField(body, "period", pos, e1);
         if(d == "" || e1 < 0)
            break;
         string sid = m_http.JsonField(body, "series", e1, e2);
         string v = m_http.JsonField(body, "value", (e2 > e1 ? e2 : e1), e3);
         pos = (e3 > e1 ? e3 : (e2 > e1 ? e2 : e1));
         datetime dt = StringToTime(d);           // accepts yyyy-mm-dd
         if(dt <= 0 || sid == "" || v == "")
            continue;
         if(s.rows > 0 && dt < s.date[s.rows - 1])
            continue;
         if(s.rows == 0 || dt > s.date[s.rows - 1])
           {
            int n = AltSeriesAppend(s, dt, dt + 6 * 86400);
            added++;
           }
         int r = s.rows - 1;
         if(sid == "WCESTUS1")      s.v1[r] = StringToDouble(v);
         else if(sid == "WCRFPUS2") s.v2[r] = StringToDouble(v);
         else if(sid == "WPULEUS3") s.v3[r] = StringToDouble(v);
        }
      if(added > 0)
        {
         AltRawSave("raw_EIA_WPSR.csv", s, 3, "date;published;stocks;production;utilization");
         PrintFormat("AltDataFetch: raw_EIA_WPSR.csv +%d weekly reports (through %s)", added,
                     TimeToString(s.date[s.rows - 1], TIME_DATE));
        }
      return added > 0;
     }

   //--- ------------------------------------ GEX forward recorder WHY THIS IS A RECORDER AND NOT A
   //--- FEATURE: option open interest is a SNAPSHOT source. Revisit when the file holds a year or
   //--- so, then screen it like every other candidate.
   string            CboeTicker(string canonical)
     {
      if(canonical == "SP500")  return "_SPX";   // index options - where dealer positioning concentrates
      if(canonical == "NAS100") return "_NDX";
      if(canonical == "US30")   return "_DJX";
      if(canonical == "US2000") return "IWM";
      if(canonical == "XAUUSD") return "GLD";
      if(canonical == "XAGUSD") return "SLV";
      if(canonical == "XTIUSD") return "USO";
      if(canonical == "BTCUSD") return "IBIT";
      return "";                                 // no listed-options proxy worth recording
     }

   //--- last date already in the file, or 0
   datetime          LastGexDate(string file)
     {
      int h = FileOpen(file, FILE_READ | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
      if(h == INVALID_HANDLE)
         return 0;
      string last = "";
      while(!FileIsEnding(h))
        {
         string line = FileReadString(h);
         if(StringLen(line) > 8 && StringFind(line, "date;") != 0)
            last = line;
        }
      FileClose(h);
      if(last == "")
         return 0;
      string f[];
      if(StringSplit(last, ';', f) < 1)
         return 0;
      return StringToTime(f[0]);
     }

   bool              UpdateGex(string canonical)
     {
      string ticker = CboeTicker(canonical);
      if(ticker == "")
         return false;
      //--- Once per UTC day, after the US cash close (21:00 UTC covers both DST regimes),
      //--- and never on a weekend - the chain would just repeat Friday's settled numbers.
      MqlDateTime g;
      TimeToStruct(TimeGMT(), g);
      if(g.day_of_week == 0 || g.day_of_week == 6)
         return false;
      if(g.hour < 21)
         return false;
      string file = ALTFETCH_DIR + "gex_" + canonical + ".csv";
      datetime today = StringToTime(StringFormat("%04d.%02d.%02d", g.year, g.mon, g.day));
      if(!m_gexDateLoaded)
        {
         m_lastGexDate = LastGexDate(file);
         m_gexDateLoaded = true;
        }
      if(m_lastGexDate >= today)
         return false;
      //--- Re-read from DISK before spending the download: two charts of the same symbol
      //--- (e.g. D1 + H4) each run their own fetcher, and a per-instance cached date would
      //--- let the second chart append a duplicate row for the day the first just recorded.
      m_lastGexDate = LastGexDate(file);
      if(m_lastGexDate >= today)
         return false;
      //--- the SPX chain is ~13 MB: this needs a far longer timeout than the small feeds
      string body;
      if(!m_http.HttpGet("https://cdn.cboe.com/api/global/delayed_quotes/options/" + ticker + ".json",
                  body, ALTFETCH_GEX_TIMEOUT_MS))
         return false;
      if(StringLen(body) < 1000)
        {
         Print("AltDataFetch: GEX response for " + ticker + " was too short to be a chain - skipping.");
         return false;
        }
      int e0;
      string spotS = m_http.JsonField(body, "current_price", 0, e0);   // absent from option rows, so
      if(spotS == "")                                           // a plain scan finds the header one
         spotS = m_http.JsonField(body, "close", 0, e0);
      double spot = StringToDouble(spotS);
      if(spot <= 0.0)
        {
         Print("AltDataFetch: GEX " + ticker + " - no usable spot price in the response, skipping.");
         return false;
        }
      //--- per-expiry accumulation (54 expiries on SPX; the cap is generous)
      int    expCode[128];
      double expGex[128];
      int    nExp = 0;
      double gexCall = 0.0, gexPut = 0.0;
      double oiCall = 0.0, oiPut = 0.0;
      int    rows = 0;
      int    pos = 0;
      //--- Row layout is option, ..., open_interest, volume, delta, gamma - so one forward
      //--- sweep reads each row's three fields in the order they appear.
      while(true)
        {
         int e1;
         string sym = m_http.JsonField(body, "option", pos, e1);
         if(sym == "" || e1 < 0)
            break;
         int e2, e3;
         string oiS = m_http.JsonField(body, "open_interest", e1, e2);
         string gS  = m_http.JsonField(body, "gamma", (e2 > e1 ? e2 : e1), e3);
         pos = (e3 > e1 ? e3 : (e2 > e1 ? e2 : e1));
         double oi = StringToDouble(oiS);
         double gam = StringToDouble(gS);
         if(oi <= 0.0 || gam == 0.0)
            continue;                        // no exposure - skip the string work entirely
         //--- OCC symbol: <root><YYMMDD><C|P><strike*1000, 8 digits>, so the tail is fixed
         //--- width regardless of how long the root is (SPX, SPXW, IWM, ...).
         int len = StringLen(sym);
         if(len < 15)
            continue;
         string kind = StringSubstr(sym, len - 9, 1);
         int expiry = (int)StringToInteger(StringSubstr(sym, len - 15, 6));
         //--- dollar gamma per 1% move: gamma x OI x 100 (contract multiplier) x S^2 x 0.01
         double dollar = gam * oi * 100.0 * spot * spot * 0.01;
         if(kind == "C")
           {
            gexCall += dollar;
            oiCall  += oi;
           }
         else
           {
            gexPut -= dollar;               // dealers short puts: opposite sign convention
            oiPut  += oi;
            dollar = -dollar;
           }
         int slot = -1;
         for(int i = 0; i < nExp; i++)
            if(expCode[i] == expiry)
              {
               slot = i;
               break;
              }
         if(slot < 0 && nExp < 128)
           {
            slot = nExp++;
            expCode[slot] = expiry;
            expGex[slot]  = 0.0;
           }
         if(slot >= 0)
            expGex[slot] += dollar;
         rows++;
        }
      if(rows == 0)
        {
         Print("AltDataFetch: GEX " + ticker + " - parsed 0 contracts with exposure, skipping "
               "(the CDN response format may have changed).");
         return false;
        }
      //--- Partial-chain guard: a truncated 13 MB download parses cleanly and yields a
      //--- plausible-but-wrong GEX. A day-over-day contract-count collapse (>50%) is the
      //--- fingerprint of truncation, not of markets - OI does not halve overnight.
      int prevRows = LastGexRowCount(file);
      if(prevRows > 0 && rows < prevRows / 2)
        {
         PrintFormat("AltDataFetch: GEX %s - only %d contracts vs %d yesterday (<50%%): response "
                     "looks TRUNCATED, not recording. Will retry on the next timer pass.",
                     ticker, rows, prevRows);
         return false;
        }
      //--- three nearest expiries, by selection (nExp is tiny)
      double front[3] = {0.0, 0.0, 0.0};
      int    frontCode = 0, prevCode = 0;
      for(int k = 0; k < 3; k++)
        {
         int best = -1;
         for(int i = 0; i < nExp; i++)
           {
            if(expCode[i] <= prevCode)
               continue;
            if(best < 0 || expCode[i] < expCode[best])
               best = i;
           }
         if(best < 0)
            break;
         front[k] = expGex[best] / 1e9;
         prevCode = expCode[best];
         if(k == 0)
            frontCode = prevCode;
        }
      bool isNew = !FileIsExist(file, FILE_COMMON);
      int h = FileOpen(file, FILE_READ | FILE_WRITE | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
      if(h == INVALID_HANDLE)
        {
         PrintFormat("AltDataFetch: cannot write %s (%d) - GEX not recorded today.", file, GetLastError());
         return false;
        }
      if(isNew)
         FileWriteString(h, "date;recorded_utc;spot;gex_total;gex_call;gex_put;oi_call;oi_put;"
                            "gex_exp1;gex_exp2;gex_exp3;expiry1;n_expiry;n_rows\n");
      FileSeek(h, 0, SEEK_END);
      FileWriteString(h, StringFormat("%s;%s;%.4f;%.6f;%.6f;%.6f;%.0f;%.0f;%.6f;%.6f;%.6f;%d;%d;%d\n",
                                      TimeToString(today, TIME_DATE),
                                      TimeToString(TimeGMT(), TIME_DATE | TIME_MINUTES),
                                      spot, (gexCall + gexPut) / 1e9, gexCall / 1e9, gexPut / 1e9,
                                      oiCall, oiPut, front[0], front[1], front[2],
                                      frontCode, nExp, rows));
      FileClose(h);
      m_lastGexDate = today;
      //--- Progress matters here: the file is useless until it is long enough to screen, and
      //--- the only way to know how far along it is, is to say so.
      int have = 1;
      datetime probe = LastGexDate(file);
      if(probe > 0)
         have = CountGexRows(file);
      PrintFormat("AltDataFetch: GEX %s recorded - total %+.2f Bn/1%% (calls %+.2f, puts %+.2f), "
                  "%d contracts, %d expiries, spot %.2f. %d day(s) banked; this is a FORWARD "
                  "RECORDER - no free history exists, so it feeds no model until there is enough "
                  "to screen (~250 trading days).",
                  ticker, (gexCall + gexPut) / 1e9, gexCall / 1e9, gexPut / 1e9, rows, nExp, spot, have);
      return true;
     }

   //--- n_rows (field 14) of the last data line - the previous session's contract count
   int               LastGexRowCount(string file)
     {
      int h = FileOpen(file, FILE_READ | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
      if(h == INVALID_HANDLE)
         return 0;
      string last = "";
      while(!FileIsEnding(h))
        {
         string line = FileReadString(h);
         if(StringLen(line) > 8 && StringFind(line, "date;") != 0)
            last = line;
        }
      FileClose(h);
      if(last == "")
         return 0;
      string f[];
      if(StringSplit(last, ';', f) < 14)
         return 0;
      return (int)StringToInteger(f[13]);
     }

   int               CountGexRows(string file)
     {
      int h = FileOpen(file, FILE_READ | FILE_TXT | FILE_ANSI | FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_COMMON);
      if(h == INVALID_HANDLE)
         return 0;
      int n = 0;
      while(!FileIsEnding(h))
        {
         string line = FileReadString(h);
         if(StringLen(line) > 8 && StringFind(line, "date;") != 0)
            n++;
        }
      FileClose(h);
      return n;
     }

   //--- -------------------------------- feature math (mirrors export.py)
   //--- pandas rolling(W, min_periods=M).rank(pct=True) on the LAST element: rank of the current
   //--- value among the window's values (average rank for ties) divided by the window count.
   double            RollingPctRank(const double &v[], int i, int window, int minPeriods)
     {
      int lo = MathMax(0, i - window + 1);
      int n = i - lo + 1;
      if(n < minPeriods)
         return EMPTY_VALUE;
      double cur = v[i];
      int below = 0, ties = 0;
      for(int k = lo; k <= i; k++)
        {
         if(v[k] < cur)
            below++;
         else if(v[k] == cur)
            ties++;
        }
      double rank = below + (1.0 + ties) / 2.0;
      return rank / n;
     }

   //--- as-of index: largest row with published <= day, else -1
   int               AsOf(const SAltRawSeries &s, datetime day)
     {
      int lo = 0, hi = s.rows - 1, ans = -1;
      while(lo <= hi)
        {
         int mid = (lo + hi) / 2;
         if(s.published[mid] <= day)
           {
            ans = mid;
            lo = mid + 1;
           }
         else
            hi = mid - 1;
        }
      return ans;
     }

   //--- raw series kept resident between timer ticks (one chart = one symbol, so only the
   //--- sources the chart's catalog row names are ever populated)
   //--- m_ivol holds whichever instrument-specific implied-vol index THIS chart's catalog
   //--- row names (GVZ for gold, OVX for oil, VXN/VXD/RVX for the index symbols) - one
   //--- chart is one symbol, so a single slot serves regardless of which series it is.
   SAltRawSeries     m_vix, m_usd, m_cot, m_eia, m_ivol;
   //--- US macro block. UNRATE is the exception: it IS revised (seasonal refits, ~0.1-0.2pp), there
   //--- is no unrevised variant, and the EA cannot run the ALFRED first-print protocol - so a fresh
   //--- backfill carries small revisions into history.
   SAltRawSeries     m_dgs10, m_curve, m_bei, m_dff, m_ecb, m_cpi, m_unrate;

   //--- generic HTTP client and symbol catalog/user-mapping are separate collaborators - see
   //--- AltDataHttpClient.mqh/AltDataCatalog.mqh. Plain composition (no view/adapter): this class
   //--- has no single-inheritance parent forcing one, unlike the Expert\AIBase\* partials.
   CAltDataHttpClient m_http;
   CAltDataCatalog   m_catalog;

public:
                     CAltDataFetch(void) : m_fredKey(""), m_keyLoaded(false),
                                           m_eiaKey(""), m_eiaWarned(false),
                                           m_lastGexDate(0), m_gexDateLoaded(false)
     {
      for(int i = 0; i < ArraySize(m_lastAttempt); i++)
         m_lastAttempt[i] = 0;
     }

   //--- ------------------------------------- symbol mapping interface (forwards to the catalog)
   bool              NeedsMapping(string symbol) { return m_catalog.NeedsMapping(symbol); }
   int               CatalogCount(void) { return m_catalog.CatalogCount(); }
   string            CatalogName(int i) { return m_catalog.CatalogName(i); }
   string            CatalogLabel(int i) { return m_catalog.CatalogLabel(i); }
   bool              SaveUserMapping(string symbol, string canonical)
     {
      return m_catalog.SaveUserMapping(symbol, canonical);
     }

   //--- Maintain the raw caches and the feature CSV for `symbol`. Returns true when the feature
   //--- file was rebuilt (caller should reload the panels).
   bool              Update(string symbol)
     {
      int si = m_catalog.FindSpec(symbol);
      if(si < 0)
         return false;   // unknown symbol or user declined - nothing to serve
      SAltSymbolSpec spec = m_catalog.Spec(si);
      bool needVix = StringFind(spec.features, "vix") >= 0;
      bool needUsd = StringFind(spec.features, "usd_") >= 0;
      bool needEia = StringFind(spec.features, "eia_") >= 0;
      bool needCot = (spec.cotWhere != "");
      bool needIvol = (spec.ivolSeries != "" && StringFind(spec.features, "ivol") >= 0);
      bool needMac = StringFind(spec.features, "mac_") >= 0;
      bool changed = false;
      //--- plausibility bands: generous enough for any real print in the series' history
      //--- (VIX peaked 89 in 2008; utilization-style checks live in FeatureValue), tight
      //--- enough that transport garbage cannot pass
      if(needVix)
        {
         if(m_vix.rows == 0)
            AltRawLoad("raw_VIXCLS.csv", m_vix, 1);
         if(UpdateFred("VIXCLS", m_vix, 0, 4, 1.0, 200.0))
            changed = true;
        }
      if(needUsd)
        {
         //--- staleDays 10: the H.10 dollar index posts with ~a week's lag (live fetch
         //--- 2026-08-16 was current only through 08-07), so a 4-day horizon would fire a
         //--- futile refetch attempt every hour in the gap between releases
         if(m_usd.rows == 0)
            AltRawLoad("raw_DTWEXBGS.csv", m_usd, 1);
         if(UpdateFred("DTWEXBGS", m_usd, 1, 10, 50.0, 250.0))
            changed = true;
        }
      if(needCot)
        {
         //--- cache is named by CANONICAL, so two brokers' names for the same instrument
         //--- share one download instead of fetching the same contract twice
         string cache = "raw_COT_" + spec.canonical + ".csv";
         if(m_cot.rows == 0)
            AltRawLoad(cache, m_cot, 3);
         if(UpdateCot(cache, spec.cotWhere, m_cot, 2))
            changed = true;
        }
      if(needEia)
        {
         if(m_eia.rows == 0)
            AltRawLoad("raw_EIA_WPSR.csv", m_eia, 3);
         if(UpdateEia(m_eia, 3))
            changed = true;
        }
      if(needIvol)
        {
         if(m_ivol.rows == 0)
            AltRawLoad("raw_" + spec.ivolSeries + ".csv", m_ivol, 1);
         if(UpdateFred(spec.ivolSeries, m_ivol, 4, 4, 1.0, 300.0))
            changed = true;
        }
      if(needMac)
        {
         if(m_dgs10.rows == 0)  AltRawLoad("raw_DGS10.csv", m_dgs10, 1);
         if(m_curve.rows == 0)  AltRawLoad("raw_T10Y2Y.csv", m_curve, 1);
         if(m_bei.rows == 0)    AltRawLoad("raw_T5YIE.csv", m_bei, 1);
         if(m_dff.rows == 0)    AltRawLoad("raw_DFF.csv", m_dff, 1);
         if(m_ecb.rows == 0)    AltRawLoad("raw_ECBDFR.csv", m_ecb, 1);
         if(m_cpi.rows == 0)    AltRawLoad("raw_CPIAUCNS.csv", m_cpi, 1);
         if(m_unrate.rows == 0) AltRawLoad("raw_UNRATE.csv", m_unrate, 1);
         if(UpdateFred("DGS10", m_dgs10, 5, 4, -5.0, 30.0))      changed = true;
         if(UpdateFred("T10Y2Y", m_curve, 6, 4, -10.0, 10.0))    changed = true;
         if(UpdateFred("T5YIE", m_bei, 7, 4, -5.0, 15.0))        changed = true;
         if(UpdateFred("DFF", m_dff, 8, 4, -2.0, 30.0))          changed = true;
         if(UpdateFred("ECBDFR", m_ecb, 9, 4, -5.0, 30.0))       changed = true;
         if(UpdateFred("CPIAUCNS", m_cpi, 10, 32, 20.0, 1000.0)) changed = true;
         if(UpdateFred("UNRATE", m_unrate, 11, 32, 0.0, 35.0))   changed = true;
        }
      //--- Rebuild when stale/missing and ANY needed source has rows: a temporarily dead source
      //--- leaves its columns empty (header intact) and the panel 0-fills them - degradation must
      //--- never take the working features down with it (the silent-FRED incident took exactly that
      //--- shape under the old all-sources-required gate).
      UpdateGex(spec.canonical);
      bool anyReady = (needVix && m_vix.rows > 0) || (needUsd && m_usd.rows > 0)
                      || (needCot && m_cot.rows > 0) || (needEia && m_eia.rows > 0)
                      || (needIvol && m_ivol.rows > 0) || (needMac && m_dgs10.rows > 0);
      string outFile = AltDataFileSymbol(symbol) + "_D1.csv";   // sanitizer shared with the panel
      if((changed || !FileIsExist(ALTFETCH_DIR + outFile, FILE_COMMON)) && anyReady)
         return RebuildFeatures(spec, outFile);
      return false;
     }

protected:
   //--- --------------------------------- catalog-driven feature writer Walks the daily grid and
   //--- emits exactly export.py's format; the transforms are the FIXED a-priori constants
   //--- documented there.
   bool              RebuildFeatures(const SAltSymbolSpec &spec, string outFile)
     {
      string feats[];
      int nf = StringSplit(spec.features, ';', feats);
      if(nf <= 0)
         return false;
      //--- weekly COT net-spec series with the catalog sign applied
      double cspec[];
      ArrayResize(cspec, m_cot.rows);
      for(int i = 0; i < m_cot.rows; i++)
         cspec[i] = (m_cot.v1[i] > 0) ? spec.cotSign * (m_cot.v2[i] - m_cot.v3[i]) / m_cot.v1[i] : 0.0;
      //--- EIA crude-stocks copy for the rolling percentile
      double estk[];
      ArrayResize(estk, m_eia.rows);
      for(int i = 0; i < m_eia.rows; i++)
         estk[i] = m_eia.v1[i];
      //--- temp + swap via AtomicFile.mqh, same reasoning as AltRawSave: panels on other charts read this file
      string tmpName = "";
      int h = AtomicWriteBegin(ALTFETCH_DIR + outFile, FILE_COMMON, tmpName, FILE_TXT | FILE_ANSI);
      if(h == INVALID_HANDLE)
         return false;
      string header = "date";
      for(int f = 0; f < nf; f++)
         header += ";" + feats[f];
      FileWriteString(h, header + "\n");
      datetime today = TimeCurrent();
      for(datetime day = ALTFETCH_GRID_START; day <= today; day += 86400)
        {
         string line = TimeToString(day, TIME_DATE);
         bool any = false;
         for(int f = 0; f < nf; f++)
           {
            string val = FeatureValue(feats[f], day, cspec, estk);
            if(val != "")
               any = true;
            line += ";" + val;
           }
         if(any)
            FileWriteString(h, line + "\n");
        }
      if(!AtomicWriteEnd(h, ALTFETCH_DIR + outFile, tmpName, FILE_COMMON, true, __FUNCTION__))
         return false;
      PrintFormat("AltDataFetch: %s rebuilt (EA-maintained) - %d features: %s.",
                  outFile, nf, spec.features);
      return true;
     }

   //--- One feature value at one day; "" while its source history is not yet deep enough.
   //--- Each branch does its own as-of lookup (binary search, cheap at rebuild frequency),
   //--- so adding a source never widens a parameter list again.
   string            FeatureValue(string id, datetime day,
                                  const double &cspec[], const double &estk[])
     {
      if(id == "vix_chg5")
        {
         int i = AsOf(m_vix, day);
         return (i >= 5) ? DoubleToString((m_vix.v1[i] - m_vix.v1[i - 5]) / 10.0, 6) : "";
        }
      if(id == "vix")
        {
         int i = AsOf(m_vix, day);
         return (i >= 0) ? DoubleToString(m_vix.v1[i] / 100.0, 6) : "";
        }
      //--- the chart's OWN implied-vol index; same a-priori scales as vix/vix_chg5 so the
      //--- pairs are directly comparable to the first batch-norm layer
      if(id == "ivol_chg5")
        {
         int i = AsOf(m_ivol, day);
         return (i >= 5) ? DoubleToString((m_ivol.v1[i] - m_ivol.v1[i - 5]) / 10.0, 6) : "";
        }
      if(id == "ivol")
        {
         int i = AsOf(m_ivol, day);
         return (i >= 0) ? DoubleToString(m_ivol.v1[i] / 100.0, 6) : "";
        }
      if(id == "usd_chg5")
        {
         int i = AsOf(m_usd, day);
         return (i >= 5) ? DoubleToString(m_usd.v1[i] - m_usd.v1[i - 5], 6) : "";
        }
      if(id == "cot_spec_net")
        {
         int i = AsOf(m_cot, day);
         return (i >= 0 && m_cot.v1[i] > 0) ? DoubleToString(cspec[i], 6) : "";
        }
      if(id == "cot_idx_1y" || id == "cot_idx_3y")
        {
         int i = AsOf(m_cot, day);
         if(i < 0)
            return "";
         bool oneYear = (id == "cot_idx_1y");
         double r = RollingPctRank(cspec, i, oneYear ? 52 : 156, oneYear ? 26 : 78);
         return (r != EMPTY_VALUE) ? DoubleToString(r - 0.5, 6) : "";
        }
      if(id == "cot_chg_4w")
        {
         int i = AsOf(m_cot, day);
         return (i >= 4) ? DoubleToString(cspec[i] - cspec[i - 4], 6) : "";
        }
      if(id == "eia_stk_idx1y")
        {
         int i = AsOf(m_eia, day);
         if(i < 0 || estk[i] <= 0)
            return "";
         double r = RollingPctRank(estk, i, 52, 26);
         return (r != EMPTY_VALUE) ? DoubleToString(r - 0.5, 6) : "";
        }
      if(id == "eia_stk_chg4")
        {
         int i = AsOf(m_eia, day);
         return (i >= 4 && estk[i] > 0 && estk[i - 4] > 0)
                ? DoubleToString((estk[i] / estk[i - 4] - 1.0) * 10.0, 6) : "";
        }
      if(id == "eia_util")
        {
         int i = AsOf(m_eia, day);
         return (i >= 0 && m_eia.v3[i] > 0)
                ? DoubleToString((m_eia.v3[i] - 90.0) / 10.0, 6) : "";
        }
      //--- US macro block. Index arithmetic is in OBSERVATIONS (business days for the
      //--- daily series, months for CPI/UNRATE), matching the research screen exactly.
      if(id == "mac_y10")
        {
         int i = AsOf(m_dgs10, day);
         return (i >= 20) ? DoubleToString(m_dgs10.v1[i] - m_dgs10.v1[i - 20], 6) : "";
        }
      if(id == "mac_curve")
        {
         int i = AsOf(m_curve, day);
         return (i >= 0) ? DoubleToString(m_curve.v1[i], 6) : "";
        }
      if(id == "mac_bei")
        {
         int i = AsOf(m_bei, day);
         return (i >= 20) ? DoubleToString(m_bei.v1[i] - m_bei.v1[i - 20], 6) : "";
        }
      if(id == "mac_gap")
        {
         int i = AsOf(m_dff, day), j = AsOf(m_ecb, day);
         return (i >= 0 && j >= 0)
                ? DoubleToString((m_dff.v1[i] - m_ecb.v1[j]) / 10.0, 6) : "";
        }
      if(id == "mac_cpi")
        {
         int i = AsOf(m_cpi, day);
         return (i >= 12 && m_cpi.v1[i - 12] > 0)
                ? DoubleToString((m_cpi.v1[i] / m_cpi.v1[i - 12] - 1.0) * 10.0, 6) : "";
        }
      if(id == "mac_unemp")
        {
         int i = AsOf(m_unrate, day);
         return (i >= 12) ? DoubleToString((m_unrate.v1[i] - m_unrate.v1[i - 12]) / 10.0, 6) : "";
        }
      return "";
     }
  };