Upload folder using huggingface_hub

QUALITY_SCORE_ARCHITECTURE.md

@@ -1,165 +1,13 @@
-# Token Quality / Health Score (q) - Architecture
+OK I think I see the real issue now.
 
-This document defines the "quality/health" scalar `q` used by Apollo.
+The weighted sampling balances which tokens the model sees equally. But the labels (the actual return values) are determined by the random T_cutoff within each token, not the token's class.
 
-## 1) What problem this solves
+Even a class 5 token (100x return) only pumps in a tiny window of its lifetime. If you have 1000 trades and the pump happens between trade 200-400, then:
 
-We want a single number that captures **how healthy / organic vs controlled** a token looks, so a downstream trading policy (e.g., RL agent) can treat it as a **risk/health input**.
+T_cutoff at trade 50 → returns might be +500% ✅
+T_cutoff at trade 500 → returns are -80% (post-pump bleed)
+T_cutoff at trade 700 → returns are -90%
+T_cutoff at trade 900 → returns are -95%
+So even for class 5 tokens, 80%+ of the cached training samples have negative Ground Truth labels. The model is correctly learning that at any random moment, even a "good" token is most likely going down. The class balancing doesn't change the fact that the actual Y labels are overwhelmingly negative across all classes.
 
-Key points:
-- This is **not model confidence**.
-- `q` is computed **offline** using a token's **full lifetime** (for labels / training targets).
-- At **inference**, the model predicts `q` from **partial observations**.
-- We avoid hard thresholds and raw-scale features (USD, SOL, counts) by using **within-regime distributions**.
-
-## 2) Core idea (distribution-first, not rules-first)
-
-Raw totals (fees, volume, holders) are mostly **scale** and are extremely heavy-tailed. Using them directly:
-- makes the signal unstable across regimes,
-- makes it sensitive to market-wide shifts,
-- and invites hand-tuned weights ("human bias").
-
-Instead we map each metric to a **percentile** within a comparable peer group, then aggregate.
-
-## 3) Return bucketing (why it is required)
-
-The dataset is highly imbalanced: most tokens die early (<2-3x), while a tiny tail produces 10x-1000x outcomes.
-
-If you compute percentiles globally:
-- 100x tokens will always dominate "good" percentiles for scale metrics,
-- and "quality" will collapse into "return magnitude".
-
-So we compute distributions **within return regimes**.
-
-### 3.1 Bucket definition (example)
-
-Let `R_max` be the token's lifetime max return multiple (e.g., ATH / launch).
-
-Use coarse buckets for the bulk and finer buckets for the tail, e.g.:
-- B0: `R_max < 3`
-- B1: `3 <= R_max < 5`
-- B2: `5 <= R_max < 10`
-- B3: `10 <= R_max < 20`
-- B4: `20 <= R_max < 100`
-- B5: `100 <= R_max < 10_000`
-
-Notes:
-- If a bucket has too few samples, merge with a neighbor.
-- For the extreme tail you can also replace fixed buckets with **quantile buckets** on `log(R_max)` to keep sample counts stable.
-
-Interpretation (important):
-- `q` is **relative within the bucket**.
-- The "best garbage" can have high `q` in B0.
-- A 100x token can have low `q` in B4 if it looks worst vs other 100x+ tokens.
-
-This is intentional: return and quality are different axes.
-
-## 4) Feature set and sign conventions
-
-We want `q` to increase for "healthy/organic" structure and decrease for "controlled/manipulated" structure.
-
-All features below are evaluated **within the token's return bucket**.
-
-### 4.1 Scale / activity (high is usually better within-bucket)
-
-Use log transforms for stability before percentiles:
-- `log1p(total_volume_usd)`
-- `log1p(total_fees_sol)`
-- `log1p(unique_holders)`
-- `log1p(time_to_ath_sec)` (optional; see note below)
-
-Ratio features (less pure scale):
-- `fees_per_volume = total_fees_sol / (total_volume_usd + eps)`
-- `fees_per_trade  = total_fees_sol / (n_trades + eps)` (if `n_trades` exists)
-- `holders_per_trade = unique_holders / (n_trades + eps)` (if `n_trades` exists)
-- `holders_per_volume = unique_holders / (total_volume_usd + eps)`
-
-Rationale:
-- Fees and fee-per-* help separate "real urgency / competition" from "cheap wash".
-- Holders and holders-per-* help separate broad participation from concentrated looping.
-
-### 4.2 Manipulation / control (high is worse; flip sign)
-
-These are typically "the higher, the less healthy":
-- `snipers_pct_supply_top70`
-- `bundled_pct_supply`
-- `dev_hold_pct_supply`
-- `insiders_pct_supply`
-
-We treat exceptions as rare; the model can learn edge cases from context, but the label should reflect the dominant interpretation.
-
-### 4.3 Time-to-ATH note
-
-`time_to_ath_sec` can behave differently across return buckets.
-- In high-return buckets, very short times can look like a single spike / control.
-- In low-return buckets, many tokens have near-zero times because they never move.
-
-Include it only if it improves downstream behavior; keep it **bucket-relative** either way.
-
-## 5) Turning raw metrics into a signed scalar
-
-We want a single `q` in `[-1, +1]` with direction:
-- `+1` = looks healthiest vs peers in the same return bucket
-- `-1` = looks most unhealthy vs peers in the same return bucket
-
-### 5.1 Within-bucket percentile (ECDF)
-
-For each feature value `x_i`:
-- compute percentile `p_i = ECDF_b(x_i)` using only tokens in bucket `b`
-- `p_i` is in `[0, 1]`
-
-Implementation detail:
-- Use a rank-based ECDF with a small offset to avoid exact 0/1 if desired:
-  - `p_i = (rank(x_i) - 0.5) / n`
-
-### 5.2 Signed percentile
-
-Convert to signed value:
-- `s_i = 2 * p_i - 1`  (now `s_i` is in `[-1, +1]`)
-
-If "high is bad" for that feature, flip it:
-- `s_i := -s_i`
-
-This gives direction + magnitude in a single number.
-
-### 5.3 Aggregate without hand weights
-
-To avoid hand-tuned weights, use a symmetric aggregator:
-- `q_raw = mean_i(s_i)`
-
-Optional robustness:
-- clip each `s_i` to `[-0.99, 0.99]` before averaging (limits extreme leverage)
-- use a trimmed mean (drop top/bottom k% of `s_i`) if a single metric can be noisy
-
-### 5.4 Optional: re-rank the aggregate (final calibration)
-
-If you want the final `q` to be strictly comparable across time / retrains and more uniform within bucket:
-- `q = 2 * ECDF_b(q_raw) - 1`
-
-This keeps the "relative within bucket" meaning while stabilizing scale.
-
-## 6) Training vs inference (how it is used)
-
-Offline labeling (training target):
-1) Compute `R_max` from full lifetime.
-2) Assign return bucket `b`.
-3) Compute all chosen metrics from full lifetime.
-4) Convert metrics -> signed percentiles -> `q`.
-
-Inference (model output):
-- The model only sees partial history and must predict the *final* `q` (computed above).
-- The trading policy uses predicted return signals + predicted `q` to decide position sizing / risk.
-
-## 7) Practical notes
-
-- Use `eps` (e.g., `1e-9`) in denominators to avoid divide-by-zero.
-- If a metric is missing for a token, drop it from the mean for that token (or impute with bucket median).
-- When bucket sample counts drift, prefer merging buckets rather than letting ECDF be noisy.
-- Recompute distributions on the same "source-of-truth" dataset used for training (not ad-hoc caches).
-
-## 8) Summary
-
-`q` is a **return-regime-relative**, **distribution-normalized**, **signed** health score:
-- It is not a threshold classifier.
-- It avoids raw-scale dependence and hand weighting.
-- It cleanly separates "made money" (return) from "looks healthy" (quality).
+The model isn't broken — it learned exactly what the data showed it. The issue is that the training setup doesn't teach it to recognize the pre-pump moment specifically.

data/data_loader.py

@@ -124,7 +124,6 @@ class OracleDataset(Dataset):
                  max_samples: Optional[int] = None,
 
                  token_allowlist: Optional[List[str]] = None,
-                 t_cutoff_seconds: int = 60,
                  cache_dir: Optional[Union[str, Path]] = None,
                  start_date: Optional[datetime.datetime] = None,
                  min_trade_usd: float = 0.0,
@@ -161,10 +160,7 @@ class OracleDataset(Dataset):
         
 
 
-        # If a fetcher is provided, we can determine the number of samples.
-        # Otherwise, we are likely in a test mode where __len__ might not be called
-        # or is used with a mock length.
-        self.t_cutoff_seconds = max(0, int(t_cutoff_seconds or 0))
+
         self.token_allowlist = set(token_allowlist) if token_allowlist else None
         
         if self.cache_dir:
@@ -2606,7 +2602,7 @@ class OracleDataset(Dataset):
                     pooler.pool_map[key] = {'item': emb.cpu().clone(), 'idx': old_entry['idx']}
 
 
-    def __cacheitem_context__(self, idx: int, num_samples_per_token: int = 1, encoder: Optional[Any] = None) -> List[Optional[Dict[str, Any]]]:
+    def __cacheitem_context__(self, idx: int, num_samples_per_token: int = 1, encoder: Optional[Any] = None, forced_cutoff_trade_idx: Optional[int] = None) -> List[Optional[Dict[str, Any]]]:
         """
         Generates fully processed training contexts for caching.
 
@@ -2957,7 +2953,14 @@ class OracleDataset(Dataset):
         results = []
 
         # Sample indices (with replacement if needed)
-        if num_samples_per_token >= len(eligible_indices):
+        if forced_cutoff_trade_idx is not None:
+            # Forced mode: use the exact trade index provided (for evaluation)
+            if forced_cutoff_trade_idx >= len(all_trades_sorted):
+                print(f"  WARN: forced_cutoff_trade_idx={forced_cutoff_trade_idx} >= total trades {len(all_trades_sorted)}, clamping.")
+                forced_cutoff_trade_idx = len(all_trades_sorted) - 2
+            sampled_indices = [forced_cutoff_trade_idx]
+            print(f"  Using forced T_cutoff at trade index {forced_cutoff_trade_idx}")
+        elif num_samples_per_token >= len(eligible_indices):
             sampled_indices = eligible_indices.copy()
         else:
             sampled_indices = random.sample(eligible_indices, num_samples_per_token)

ingest.sh

@@ -20,7 +20,7 @@ error() { echo -e "${RED}[ERROR]${NC} $1"; exit 1; }
 #===============================================================================
 header "Step 5-6/7: Processing Epochs (Download → Ingest → Delete)"
 
-EPOCHS=(844 845 846 847 848 849 850 851)
+EPOCHS=(852 853)
 
 
 log "Processing epochs one at a time to minimize disk usage..."
@@ -39,9 +39,9 @@ for epoch in "${EPOCHS[@]}"; do
         error "Failed to download epoch ${epoch}. Cannot continue."
     }
 
-    # Step 2: Ingest (always pass --merge-neo4j; auto-detect handles empty DB)
-    log "  [2/3] Ingesting epoch ${epoch} into databases..."
-    python scripts/ingest_epoch.py --epoch "$epoch" --merge-neo4j || {
+    # Step 2: Ingest (ClickHouse only)
+    log "  [2/3] Ingesting epoch ${epoch} into ClickHouse database..."
+    python scripts/ingest_epoch.py --epoch "$epoch" --skip-neo4j || {
         error "Ingestion failed for epoch ${epoch}. Cannot continue."
     }
 

log.log

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:656c6818f224b26869b5d0ae10f6b522ff7eb5c7b1b3aeb59b34c3db218338a9
-size 11360
+oid sha256:11ac013f8e91ad65475b8106a5a072dc42f67e0773ddc4a50825e316c578e0d4
+size 3472

models/multi_modal_processor.py

@@ -81,8 +81,9 @@ class MultiModalEncoder:
         is_text = isinstance(x[0], str)
 
         autocast_dtype = self.dtype if self.dtype in [torch.float16, torch.bfloat16] else None
+        device_str = self.device.type if isinstance(self.device, torch.device) else self.device
         
-        with torch.autocast(device_type=self.device, dtype=autocast_dtype, enabled=(autocast_dtype is not None)):
+        with torch.autocast(device_type=device_str, dtype=autocast_dtype, enabled=(autocast_dtype is not None)):
             try:
                 if is_text:
                     inputs = self.processor(text=x, return_tensors="pt", padding=True, truncation=True).to(self.device)

scripts/evaluate_sample.py

@@ -23,6 +23,11 @@ from models.ohlc_embedder import OHLCEmbedder
 from models.model import Oracle
 import models.vocabulary as vocab
 from train import create_balanced_split
+from dotenv import load_dotenv
+from clickhouse_driver import Client as ClickHouseClient
+from neo4j import GraphDatabase
+from data.data_fetcher import DataFetcher
+from scripts.analyze_distribution import get_return_class_map
 
 def unlog_transform(tensor):
     """Invert the log1p transform applied during training."""
@@ -32,13 +37,13 @@ def unlog_transform(tensor):
 def parse_args():
     parser = argparse.ArgumentParser()
     parser.add_argument("--checkpoint", type=str, default="checkpoints/checkpoint-90000", help="Path to checkpoint dir")
-    parser.add_argument("--cache_dir", type=str, default="/workspace/apollo/data/cache", help="Path to dataset cache")
-    parser.add_argument("--sample_idx", type=int, default=None, help="Specific sample index to evaluate")
+    parser.add_argument("--sample_idx", type=str, default=None, help="Specific sample index or Mint Address to evaluate")
     parser.add_argument("--mixed_precision", type=str, default="bf16")
     parser.add_argument("--horizons_seconds", type=int, nargs="+", default=[300, 900, 1800, 3600, 7200])
     parser.add_argument("--quantiles", type=float, nargs="+", default=[0.1, 0.5, 0.9])
-    parser.add_argument("--seed", type=int, default=42)
-    parser.add_argument("--min_horizon", type=int, default=900, help="Ensure the sampled coin has ground truth for at least this horizon (in seconds)")
+    parser.add_argument("--seed", type=int, default=None)
+    parser.add_argument("--min_class", type=int, default=5, help="Filter out tokens with return class beneath this ID (e.g., 1 for >= 3x returns)")
+    parser.add_argument("--cutoff_trade_idx", type=int, default=600, help="Force the T_cutoff at this exact trade index (e.g., 10 = right after the 10th trade)")
     return parser.parse_args()
 
 def get_latest_checkpoint(checkpoint_dir):
@@ -52,6 +57,7 @@ def get_latest_checkpoint(checkpoint_dir):
     return None
 
 def main():
+    load_dotenv()
     args = parse_args()
     
     accelerator = Accelerator(mixed_precision=args.mixed_precision)
@@ -63,76 +69,49 @@ def main():
     elif accelerator.mixed_precision == 'fp16':
         init_dtype = torch.float16
 
-    print(f"Loading cached dataset from {args.cache_dir}...")
+    print("INFO: Initializing DB Connections for LIVE evaluation...")
+    clickhouse_host = os.getenv("CLICKHOUSE_HOST", "localhost")
+    clickhouse_port = int(os.getenv("CLICKHOUSE_PORT", 9000))
+    neo4j_uri = os.getenv("NEO4J_URI", "bolt://localhost:7687")
+    neo4j_user = os.getenv("NEO4J_USER", "neo4j")
+    neo4j_password = os.getenv("NEO4J_PASSWORD", "password")
+
+    clickhouse_client = ClickHouseClient(host=clickhouse_host, port=clickhouse_port)
+    neo4j_driver = GraphDatabase.driver(neo4j_uri, auth=(neo4j_user, neo4j_password))
+    data_fetcher = DataFetcher(clickhouse_client=clickhouse_client, neo4j_driver=neo4j_driver)
+
+    print(f"Loading live dataset generator...")
+    
+    # We inject the data fetcher directly. No cache directories are used.
     dataset = OracleDataset(
-        data_fetcher=None,
+        data_fetcher=data_fetcher,
         fetcher_config=None,
         horizons_seconds=args.horizons_seconds,
         quantiles=args.quantiles,
-        max_samples=None,
-        t_cutoff_seconds=60,
-        cache_dir=args.cache_dir
+        cache_dir=None
     )
 
-    if len(dataset) == 0:
-        raise ValueError("Dataset is empty!")
-
-    # Optionally pick validation sample like in training
-    print("Creating balanced train/val split to pick a validation sample...")
-    _, val_indices, _ = create_balanced_split(dataset, n_val_per_class=10, seed=args.seed)
-
-    # Re-seed with system time so we don't pick the same sample every time
-    import time
-    random.seed(time.time())
-
-    # --- Filter by minimum horizon if requested ---
-    if args.min_horizon is not None and args.min_horizon in args.horizons_seconds:
-        print(f"Filtering dataset to find samples with ground truth >= {args.min_horizon}s...")
-        h_idx = args.horizons_seconds.index(args.min_horizon)
-        num_quantiles = len(args.quantiles)
-        
-        valid_indices = []
-        # We search through a shuffled subset to avoid checking the whole dataset
-        search_pool = val_indices.copy()
-        random.shuffle(search_pool)
-        if not search_pool:
-            search_pool = list(range(len(dataset)))
-            random.shuffle(search_pool)
-            
-        for idx in search_pool:
-            sample = dataset[idx]
-            if sample is None:
-                continue
-            mask = sample.get('labels_mask')
-            if mask is not None:
-                # Based on raw file inspection, mask is shape [H], so we index by h_idx directly
-                if h_idx < len(mask) and mask[h_idx] > 0.0:
-                    valid_indices.append(idx)
-                    # Once we find a handful of valid ones, we can stop searching
-                    if len(valid_indices) >= 10:
-                        break
-        
-        if valid_indices:
-            print(f"Found {len(valid_indices)} candidate samples with >= {args.min_horizon}s horizon.")
-            val_indices = valid_indices
-        else:
-            print(f"WARNING: No samples found with ground truth for horizon {args.min_horizon}s. Reverting to random pick.")
+    # Filter out manipulated/broken tokens and optionally enforce min_class
+    from models.vocabulary import MANIPULATED_CLASS_ID
+    print("INFO: Fetching Return Classification Map...")
+    return_class_map, _ = get_return_class_map(clickhouse_client)
+    
+    min_class_thresh = args.min_class if args.min_class is not None else 0
+    
+    original_len = len(dataset.sampled_mints)
+    dataset.sampled_mints = [
+         m for m in dataset.sampled_mints 
+         if return_class_map.get(m['mint_address']) is not None 
+         and return_class_map.get(m['mint_address']) != MANIPULATED_CLASS_ID
+         and return_class_map.get(m['mint_address']) >= min_class_thresh
+    ]
+    dataset.num_samples = len(dataset.sampled_mints)
+    print(f"INFO: Filtered tokens. {original_len} -> {len(dataset.sampled_mints)} valid tokens (class >= {min_class_thresh}).")
 
-    if args.sample_idx is not None:
-        if args.sample_idx >= len(dataset):
-             raise ValueError(f"Sample index {args.sample_idx} out of range [0, {len(dataset)-1}]")
-        sample_idx = args.sample_idx
-    else:
-        # Pick a random sample from validation set
-        if len(val_indices) > 0:
-             sample_idx = random.choice(val_indices)
-        else:
-             print("No validation indices found. Picking random sample from entire set.")
-             sample_idx = random.randint(0, len(dataset) - 1)
-             
-    print(f"\nEvaluating on sample index: {sample_idx}")
+    if len(dataset) == 0:
+        raise ValueError("Dataset is empty. Are ClickHouse data and trade pipelines populated? (Check if min_return filtered everything out)")
 
-    # Initialize encoders and model
+    # Initialize encoders and model FIRST because we need multi_modal_encoder to compile context
     print("Initializing encoders...")
     multi_modal_encoder = MultiModalEncoder(dtype=init_dtype, device=device)
     time_encoder = ContextualTimeEncoder(dtype=init_dtype)
@@ -181,9 +160,6 @@ def main():
         model = accelerator.prepare(model)
     else:
         print(f"Loading checkpoint from {ckpt_path}...")
-        # Since we use accelerate, the state dict is usually split or in pytorch_model.bin/model.safetensors
-        # Using accelerate to load:
-        # We need to wrap it if we want to use `accelerator.load_state`
         model = accelerator.prepare(model)
         try:
             accelerator.load_state(ckpt_path)
@@ -202,7 +178,6 @@ def main():
                 else:
                     state_dict = torch.load(model_file, map_location="cpu")
                 
-                # Unwrap model to load state
                 uw_model = accelerator.unwrap_model(model)
                 uw_model.load_state_dict(state_dict, strict=False)
                 print("Successfully loaded weights directly.")
@@ -211,11 +186,51 @@ def main():
 
     model.eval()
 
-    # Get sample
-    raw_sample = dataset[sample_idx]
-    if raw_sample is None:
-        print("Sample is None!")
-        return
+    # Find a valid sample
+    valid_context_found = False
+    max_retries = 20
+    retries = 0
+    raw_sample = None
+    sample_mint_addr = None
+    
+    while not valid_context_found and retries < max_retries:
+        if args.sample_idx is not None:
+             if isinstance(args.sample_idx, str) and not args.sample_idx.isdigit():
+                 found_idx = next((i for i, m in enumerate(dataset.sampled_mints) if m['mint_address'] == args.sample_idx), None)
+                 if found_idx is None:
+                     import datetime
+                     dataset.sampled_mints.append({'mint_address': args.sample_idx, 'creator_address': '', 'timestamp': datetime.datetime.now(datetime.timezone.utc)})
+                     sample_idx = len(dataset.sampled_mints) - 1
+                 else:
+                     sample_idx = found_idx
+             else:
+                 sample_idx = int(args.sample_idx)
+                 if sample_idx >= len(dataset):
+                      raise ValueError(f"Sample index {sample_idx} out of range")
+        else:
+             sample_idx = random.randint(0, len(dataset.sampled_mints) - 1)
+             
+        sample_mint_addr = dataset.sampled_mints[sample_idx]['mint_address']
+        print(f"Trying Token Address: {sample_mint_addr}")
+        
+        contexts = dataset.__cacheitem_context__(sample_idx, num_samples_per_token=1, encoder=multi_modal_encoder, forced_cutoff_trade_idx=args.cutoff_trade_idx)
+        
+        if not contexts or len(contexts) == 0 or contexts[0] is None:
+             print("  [Failed to generate valid context pattern, skipping...]")
+             retries += 1
+             if args.sample_idx is not None:
+                  print("Specific sample requested but failed to generate context. Exiting.")
+                  return
+             continue
+             
+        raw_sample = contexts[0]
+        valid_context_found = True
+
+    if not valid_context_found:
+         print(f"Could not find a valid context after {max_retries} attempts.")
+         return
+         
+    print(f"\nEvaluating precisely on Token Address: {sample_mint_addr}")
 
     batch = collator([raw_sample])
     

status

@@ -1,3 +1,3 @@
-PID: 7085
-Started at: 2026-01-26 07:54:26
-Revision: 54508
+PID: 5825
+Started at: 2026-03-06 06:22:33
+Revision: 54510

wget-log.1

@@ -0,0 +1,11 @@
-                                                         0%[                                                                                                                              ]       0  --.-KB/s               
-                                                       100%[=============================================================================================================================>]   3.81K  --.-KB/s    in 0s      
+--2026-03-06 06:21:59--  https://debian.neo4j.com/neotechnology.gpg.key
+Resolving debian.neo4j.com (debian.neo4j.com)... 3.169.221.47, 3.169.221.34, 3.169.221.16, ...
+Connecting to debian.neo4j.com (debian.neo4j.com)|3.169.221.47|:443... connected.
+HTTP request sent, awaiting response... 200 OK
+Length: 3905 (3.8K) [application/pgp-keys]
+Saving to: ‘STDOUT’
+
+
+
+2026-03-06 06:21:59 (9.00 MB/s) - written to stdout [3905/3905]
+