// Live FinData integration tests — opt-in: RAID_LIVE=1 and FINDATA_API_KEY (or FINDATA_API_KEY_FILE) must be set.
// Each test makes a handful of small requests (well under the 600/min limit) and never writes vendor data into git
// (cache under $RAID_CACHE_DIR or datasets/cache, which is git-ignored).
#include <catch2/catch_test_macros.hpp>
#include <catch2/matchers/catch_matchers_floating_point.hpp>
#include "raid/data/findata.hpp"
#include "raid/data/io.hpp"
#include "raid/data/resample.hpp"
#include "raid/data/returns.hpp"
#include "raid/estimators/realized.hpp"
#include <cstdlib>
#include <cmath>
using namespace raid; using namespace raid::findata; using Catch::Matchers::WithinRel;

static bool live() { const char* e = std::getenv("RAID_LIVE"); return e && std::string(e) == "1"; }

TEST_CASE("Live: health, catalog and provenance headers", "[live][findata]") {
  if (!live()) { SKIP("RAID_LIVE != 1"); }
  Client c; auto h = c.health(); CHECK(h["status"] == "ok");
  auto cat = c.catalog_datasets(); bool found = false; for (auto& d : cat["data"]) if (d["id"] == "market_bars.stock.1m.raw") found = true; CHECK(found);
  auto s = c.sessions("SPY", Interval::m1, date_from_iso("2024-11-25"), date_from_iso("2024-11-29"));
  usize early = 0; for (auto& r : s["data"]) if (r["session_date"] == "2024-11-29") { early = r["expected_bars"].get<usize>(); } CHECK(early == 211);
  CHECK_FALSE(c.log().empty()); CHECK_FALSE(c.log().back().request_id.empty());
}

TEST_CASE("Live: bars pages equal export rows; calendar agrees with FinData sessions; aggregate matches derived 5m", "[live][findata]") {
  if (!live()) { SKIP("RAID_LIVE != 1"); }
  Client c; BarsRequest br; br.symbol = "AAPL"; br.start = date_from_iso("2024-11-25"); br.end = date_from_iso("2024-11-29");
  auto b = c.bars(br); CHECK(b.bars.n_sessions() == 4); CHECK(b.dataset_version > 0); CHECK(b.dataset_id == "market_bars.stock.1m.raw");
  MarketCalendar cal; auto q = b.bars.validate(cal); CHECK_FALSE(q.has_errors()); CHECK(q.coverage > 0.99);
  auto sess = c.sessions("AAPL", Interval::m1, br.start, br.end);
  for (auto& r : sess["data"]) { const Date d = date_from_iso(r["session_date"].get<std::string>()); CHECK(cal.expected_regular_bars(d, Interval::m1) == r["expected_bars"].get<usize>());
    const usize k = static_cast<usize>(std::find(b.bars.session_dates.begin(), b.bars.session_dates.end(), d) - b.bars.session_dates.begin()); REQUIRE(k < b.bars.n_sessions());
    CHECK(static_cast<usize>(b.bars.session_offsets[k + 1] - b.bars.session_offsets[k]) == r["bar_count"].get<usize>()); }
  // second call served from cache
  auto b2 = c.bars(br); CHECK(b2.from_cache); CHECK(b2.bars.size() == b.bars.size());
  // export of the same window
  ExportRequest er; er.dataset_id = "market_bars.stock.1m.raw"; er.symbols = {"AAPL"}; er.start = br.start; er.end = br.end;
  auto e = c.export_bars(er); CHECK(e.rows == b.bars.size()); CHECK(e.sha256.size() == 64); CHECK(e.dataset_version == b.dataset_version);
#ifdef RAID_WITH_ARROW
  auto pq = data::read_bars_parquet(e.path); REQUIRE(pq.series.count("AAPL")); CHECK(pq.series["AAPL"].size() == b.bars.size());
  for (usize i = 0; i < b.bars.size(); i += 97) CHECK(pq.series["AAPL"].close[i] == b.bars.close[i]);
  // calendar-aware aggregation vs FinData derived 5m (regular session)
  auto reg = b.bars.filter_session(cal, SessionPolicy::regular); auto agg = data::aggregate(reg, Interval::m5, cal);
  BarsRequest b5; b5.symbol = "AAPL"; b5.interval = Interval::m5; b5.start = br.start; b5.end = br.end; auto five = c.bars(b5);
  auto five_reg = five.bars.filter_session(cal, SessionPolicy::regular);
  INFO("agg bars " << agg.size() << " findata 5m regular bars " << five_reg.size());
  usize matched = 0, mism = 0; for (usize i = 0; i < agg.size(); ++i) { auto it = std::lower_bound(five_reg.t_open.begin(), five_reg.t_open.end(), agg.t_open[i]); if (it == five_reg.t_open.end() || *it != agg.t_open[i]) continue; const usize j = static_cast<usize>(it - five_reg.t_open.begin());
    const bool ok = agg.open[i] == five_reg.open[j] && agg.high[i] == five_reg.high[j] && agg.low[i] == five_reg.low[j] && agg.close[i] == five_reg.close[j] && std::abs(agg.volume[i] - five_reg.volume[j]) < 1e-6; ok ? ++matched : ++mism; if (!ok && mism <= 4) { std::fprintf(stderr, "[live] mismatch %s o %.4f/%.4f h %.4f/%.4f l %.4f/%.4f c %.4f/%.4f v %.0f/%.0f\n", timestamp_iso(agg.t_open[i]).c_str(), agg.open[i], five_reg.open[j], agg.high[i], five_reg.high[j], agg.low[i], five_reg.low[j], agg.close[i], five_reg.close[j], agg.volume[i], five_reg.volume[j]); UNSCOPED_INFO("mismatch at " << timestamp_iso(agg.t_open[i]) << " o " << agg.open[i] << "/" << five_reg.open[j] << " c " << agg.close[i] << "/" << five_reg.close[j] << " v " << agg.volume[i] << "/" << five_reg.volume[j]); } }
  std::fprintf(stderr, "[live] aggregate 1m->5m vs FinData derived 5m: matched=%zu mismatched=%zu agg_bars=%zu findata_bars=%zu\n", matched, mism, agg.size(), five_reg.size()); INFO("matched " << matched << " mismatched " << mism); CHECK(matched > 0);
  CHECK(mism <= 4);   // regular policy: RAID's closing bin = the 16:00 print only; FinData's derived 16:00 bin also contains 16:01–16:04 (post-close) — one bin per session differs
  // Under the extended policy the bins are built on all 04:00–19:59 bars and must reproduce FinData's derived 5m exactly on the regular window.
  auto agg_ext = data::aggregate(b.bars, Interval::m5, cal, SessionPolicy::extended);
  usize m2 = 0, mm2 = 0; for (usize j = 0; j < five_reg.size(); ++j) { auto it = std::lower_bound(agg_ext.t_open.begin(), agg_ext.t_open.end(), five_reg.t_open[j]); if (it == agg_ext.t_open.end() || *it != five_reg.t_open[j]) { ++mm2; continue; } const usize i = static_cast<usize>(it - agg_ext.t_open.begin());
    const bool ok = agg_ext.open[i] == five_reg.open[j] && agg_ext.high[i] == five_reg.high[j] && agg_ext.low[i] == five_reg.low[j] && agg_ext.close[i] == five_reg.close[j] && std::abs(agg_ext.volume[i] - five_reg.volume[j]) < 1e-6; ok ? ++m2 : ++mm2; }
  std::fprintf(stderr, "[live] extended-policy aggregate vs FinData 5m on the regular window: matched=%zu mismatched=%zu\n", m2, mm2);
  CHECK(mm2 == 0);
#endif
}