Please read my plan and add more to it for maximum improvement: **Project: Rust Divergence Finder**

**Goal:** To create a Rust application that ingests candle and indicator data from PostgreSQL, identifies various types of price/indicator divergences, and outputs these findings.

**Core Principles:**

1.  **Robustness:** Error handling, efficient data processing, testability, clear logging.
2.  **Flexibility:** Support for multiple indicators and configurable divergence parameters.
3.  **Performance:** Leverage Rust's strengths for efficient computation, especially with time-series data.
4.  **Maintainability:** Well-structured code, clear separation of concerns.

---

**I. Planning & Design Phase**

**A. Understanding Divergences**

*   **Regular (Classic) Divergence:**
    *   **Bullish:** Price makes Lower Lows (LL), Indicator makes Higher Lows (HL). Suggests weakening downtrend, potential reversal up.
    *   **Bearish:** Price makes Higher Highs (HH), Indicator makes Lower Highs (LH). Suggests weakening uptrend, potential reversal down.
*   **Hidden (Continuation) Divergence:**
    *   **Bullish:** Price makes Higher Lows (HL), Indicator makes Lower Lows (LL). Suggests continuation of uptrend.
    *   **Bearish:** Price makes Lower Highs (LH), Indicator makes Higher Highs (HH). Suggests continuation of downtrend.
*   **Exaggerated Divergence (Optional, less common):**
    *   Price makes double bottom/top, indicator makes a higher low / lower high. Can be treated as a weaker form of regular divergence. For now, focus on the main 4.

**B. Data Requirements**

*   **Candles:** Timestamp, Open, High, Low, Close, Volume (optional but good).
*   **Indicators:** Timestamp, Value(s). Examples:
    *   RSI: Single value.
    *   MACD: MACD line, Signal line, Histogram. Divergence can be sought on any of these.
    *   Example data structure: Candle Data:
symbol, open_time, open, high, low, close, volume, close_time, quote_asset_volume, number_of_trades, taker_buy_base_asset_volume, taker_buy_quote_asset_volume, last_updated

"BNBUSDT"   "2024-04-24 20:45:00+01"    602.10000000    605.00000000    601.90000000    604.20000000    8148.94800000   "2024-04-24 20:59:59.999+01"    4923004.92590000    3852    4762.29600000   2876784.40970000    "2025-04-24 20:35:22.2664+01"

Indicator Data:
macd:
symbol, timeframe, open_time, close_time, macd, macd_signal, macd_histogram
"BNBUSDT"   "15m"   "2024-04-25 05:00:00+01"    "2024-04-25 05:14:59.999+01"    -0.60390481842918170380372872   -0.3802403461597821936791006633 -0.2236644722693995101246280567

rsi:
symbol, timeframe, open_time, close_time, rsi
"BNBUSDT"   "15m"   "2024-04-25 00:15:00+01"    "2024-04-25 00:29:59.999+01"    58.008658008658008658008658006

    *   Volume: Can also show divergence with price.
*   **Key for Divergence Detection:** Price Highs/Lows and corresponding Indicator peaks/troughs.

**C. System Architecture**

```
+---------------------+     +----------------------+     +-----------------------+
|    PostgreSQL DB    | --> | Data Ingestion Layer | --> |  Processing Engine    |
| - Candles           |     | (Rust - SQLx)        |     |  (Rust)               |
| - Indicator Data    |     +----------------------+     |  - Peak/Trough Finder |
| - Config (optional) |                                  |  - Divergence Logic   |
+---------------------+                                  +-----------------------+
         ^                                                         |
         | (Store results)                                         |
         +-----------------------+     +-----------------------+   |
         |   Output/Reporting    | <-- |   Divergence Results  | <-+
         | - DB Table            |     |   (Rust Structs)      |
         | - Console Log         |     +-----------------------+
         | - File (CSV/JSON)     |
         +-----------------------+
```

**D. Configuration**

*   Database connection details.
*   Symbols to process (e.g., `BTCUSDT`, `ETHUSDT`).
*   Timeframes to process (e.g., `1h`, `4h`).
*   Indicator configurations:
    *   Indicator name (e.g., "RSI", "MACD_Histogram").
    *   DB table name for the indicator.
    *   DB column name for the indicator value to use.
    *   Price series to use for comparison (e.g., `Close`, `High`, `Low`).
*   Parameters for peak/trough detection (e.g., window size or prominence).
*   Parameters for divergence validation (e.g., max number of bars between extrema).
*   Output options (DB table, log level, file paths).

---

**II. Development Plan**

**A. Setup & Dependencies (Cargo.toml)**

```toml
[dependencies]
tokio = { version = "1", features = ["full"] }  # Async runtime
sqlx = { version = "0.7", features = ["runtime-tokio-rustls", "postgres", "macros", "chrono", "decimal", "json"] } # Async SQL
chrono = { version = "0.4", features = ["serde"] } # Timestamps
serde = { version = "1", features = ["derive"] } # Serialization/Deserialization
serde_json = "1" # For config if using JSON
config = "0.13" # For config file handling (supports TOML, JSON, YAML etc.)
thiserror = "1" # For custom error types
anyhow = "1"    # For simple error handling in main/higher levels
log = "0.4"
env_logger = "0.10" # Or tracing
clap = { version = "4", features = ["derive"] } # CLI arguments

# Potentially for calculations if not pre-calculated (but problem states data IS from PG)
# rust_decimal = "1" # For precise financial calculations if needed
```

**B. Core Data Structures (src/models.rs)**

```rust
use chrono::{DateTime, Utc};
use serde::Deserialize;

#[derive(Debug, Clone, sqlx::FromRow)]
pub struct Candle {
    pub timestamp: DateTime<Utc>,
    pub open: f64, // Or rust_decimal::Decimal
    pub high: f64,
    pub low: f64,
    pub close: f64,
    pub volume: Option<f64>,
    // Potentially symbol, timeframe if not passed contextually
}

// Generic way to handle different indicator structures
#[derive(Debug, Clone, sqlx::FromRow)]
pub struct IndicatorPoint {
    pub timestamp: DateTime<Utc>,
    pub value: f64, // The specific value we're checking for divergence (e.g., RSI value, MACD histogram)
    // pub symbol: String, // If fetching generically
    // pub timeframe: String, // If fetching generically
}

// Combined data point for easier processing
#[derive(Debug, Clone)]
pub struct TimeSeriesPoint {
    pub timestamp: DateTime<Utc>,
    pub price_high: f64,  // Usually candle high for bearish divergence detection
    pub price_low: f64,   // Usually candle low for bullish divergence detection
    pub price_close: f64, // Often used as a general price reference
    pub indicator_value: f64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Deserialize)]
pub enum DivergenceType {
    RegularBullish,
    RegularBearish,
    HiddenBullish,
    HiddenBearish,
}

#[derive(Debug, Clone)]
pub struct Extremum {
    pub index: usize, // Index in the TimeSeriesPoint Vec
    pub timestamp: DateTime<Utc>,
    pub value: f64,
    pub is_peak: bool, // true for peak, false for trough
}

#[derive(Debug, Clone, sqlx::FromRow)] // If storing to DB
pub struct Divergence {
    pub symbol: String,
    pub timeframe: String,
    pub indicator_name: String,
    pub divergence_type: DivergenceType, // Might need custom sqlx mapping for enum
    pub start_time: DateTime<Utc>,
    pub end_time: DateTime<Utc>,
    pub price_start: f64,
    pub price_end: f64,
    pub indicator_start: f64,
    pub indicator_end: f64,
}
```

**C. Configuration Module (src/config.rs)**

*   Define a struct mirroring your `config.toml` (or other format).
*   Use the `config` crate to load settings.

```rust
// Example config.toml structure
// [database]
// url = "postgres://user:pass@host/db"
//
// [[process_tasks]]
// symbol = "BTCUSDT"
// timeframe = "1h"
// indicators = [
//   { name = "RSI_14", table = "rsi_14_data", value_column = "rsi_value", price_source_bullish = "low", price_source_bearish = "high" },
//   { name = "MACD_Hist", table = "macd_12_26_9_data", value_column = "histogram", price_source_bullish = "low", price_source_bearish = "high" }
// ]
//
// [divergence_params]
// peak_trough_window = 5 // N bars left/right to define extremum
// max_bars_between_extrema = 60
```

**D. Database Interaction Layer (src/db.rs)**

*   Connection pool setup (`sqlx::PgPool`).
*   Functions to fetch:
    *   `fetch_candles(pool, symbol, timeframe, start_date, end_date) -> Result<Vec<Candle>, sqlx::Error>`
    *   `fetch_indicator_data(pool, indicator_config, symbol, timeframe, start_date, end_date) -> Result<Vec<IndicatorPoint>, sqlx::Error>`
        *   This function will need to dynamically build queries if `table` and `value_column` are dynamic, or have specific functions per known indicator structure.
*   Function to store results:
    *   `store_divergences(pool, divergences: &[Divergence]) -> Result<(), sqlx::Error>`

**E. Data Alignment and Preparation (Part of Processing Engine)**

1.  Fetch candles for a symbol/timeframe.
2.  Fetch indicator data for the chosen indicator(s) for the same symbol/timeframe.
3.  **Crucial Step:** Align candles and indicator data by timestamp.
    *   Use a `HashMap<DateTime<Utc>, Candle>` and `HashMap<DateTime<Utc>, IndicatorValue>` for efficient lookup.
    *   Iterate through sorted timestamps and create `TimeSeriesPoint` structs. Handle missing data points (e.g., if an indicator value is missing for a candle timestamp). Decide on a strategy: skip, interpolate (careful!), or error. Skipping is safest initially.
    *   Result: `Vec<TimeSeriesPoint>` sorted by time.

**F. Peak/Trough Detection (src/analysis/extrema.rs)**

*   Input: `&[TimeSeriesPoint]` and `window_size: usize` (config).
*   Output: `(Vec<Extremum>, Vec<Extremum>)` for (price peaks, price troughs) and `(Vec<Extremum>, Vec<Extremum>)` for (indicator peaks, indicator troughs).
*   Algorithm:
    *   A simple approach: A point `i` is a peak if `value[i]` is greater than `value[i-window_size ... i-1]` AND `value[i-1 ... i+window_size]`. Similar for troughs.
    *   Consider edge cases (start/end of data).
    *   Store index, timestamp, value, and type (peak/trough).
    *   One function for price (using `price_high` for peaks, `price_low` for troughs) and one for indicator (using `indicator_value`).

**G. Divergence Identification Logic (src/analysis/divergence.rs)**

*   Input:
    *   `Vec<Extremum>` for price peaks.
    *   `Vec<Extremum>` for price troughs.
    *   `Vec<Extremum>` for indicator peaks.
    *   `Vec<Extremum>` for indicator troughs.
    *   `max_bars_between_extrema: usize` (config).
    *   The original `Vec<TimeSeriesPoint>` (to get full candle data if needed).
*   Output: `Vec<Divergence>`.

*   **Algorithm Outline (Example for Regular Bullish):**
    1. Iterate through `price_troughs` (let current be `P_trough2`, previous be `P_trough1`).
    2. Iterate through `indicator_troughs` (let current be `I_trough2`, previous be `I_trough1`).
    3. **Conditions for Regular Bullish Divergence:**
        *   `P_trough2.value < P_trough1.value` (Lower Low in Price)
        *   `I_trough2.value > I_trough1.value` (Higher Low in Indicator)
        *   **Temporal Alignment:**
            *   `P_trough1.index` must be before `P_trough2.index`.
            *   `I_trough1.index` must be before `I_trough2.index`.
            *   The *indices* of corresponding extrema should be close: `abs(P_trough1.index - I_trough1.index) < N` and `abs(P_trough2.index - I_trough2.index) < N` (where N is a small configurable number, e.g., 2-3 bars, or ensure they are the *closest* valid extrema).
            *   The distance between the two price troughs (and indicator troughs) should be within `max_bars_between_extrema`: `(P_trough2.index - P_trough1.index) <= max_bars_between_extrema`.
        *   **No Conflicting Extrema:** There should be no lower price trough between `P_trough1` and `P_trough2`, and no lower indicator trough between `I_trough1` and `I_trough2`. (This is implicitly handled if you iterate through consecutive identified extrema, but needs careful thought if you allow skipping extrema).
    4. If all conditions met, create a `Divergence` struct.

*   Implement similar logic for:
    *   **Regular Bearish:** Price HH (`P_peak2.value > P_peak1.value`), Indicator LH (`I_peak2.value < I_peak1.value`).
    *   **Hidden Bullish:** Price HL (`P_trough2.value > P_trough1.value`), Indicator LL (`I_trough2.value < I_trough1.value`).
    *   **Hidden Bearish:** Price LH (`P_peak2.value < P_peak1.value`), Indicator HH (`I_peak2.value > I_peak1.value`).

*   **Optimization/Refinement:**
    *   Instead of nested loops over all extrema, consider iterating through price extrema and then finding corresponding indicator extrema within a reasonable window.
    *   A common approach is to iterate through the data once:
        1. Find all price peaks/troughs.
        2. Find all indicator peaks/troughs.
        3. Then, iterate (e.g., with two pointers or nested loops with constraints) through these lists of extrema to find pairs that satisfy divergence conditions.
        For example, for Regular Bullish: Iterate `i` from 0 to `price_troughs.len() - 2`. Let `pt1 = price_troughs[i]`, `pt2 = price_troughs[i+1]`.
        Then search for `it1` and `it2` in `indicator_troughs` such that `it1` is "close" to `pt1` and `it2` is "close" to `pt2` in time, and they satisfy the value conditions.

**H. Main Application Logic (src/main.rs)**

1.  Initialize logging (`env_logger::init()`).
2.  Parse CLI arguments (`clap`). (e.g., config file path, specific symbol/timeframe override).
3.  Load configuration.
4.  Establish database connection pool.
5.  Loop through configured symbols and timeframes (potentially concurrently using `tokio::spawn` and `futures::future::join_all`).
    *   For each symbol/timeframe/indicator configuration:
        a.  Fetch candles.
        b.  Fetch corresponding indicator data.
        c.  Align data into `Vec<TimeSeriesPoint>`. Handle errors if data is insufficient or misaligned.
        d.  Detect price extrema (highs for bearish, lows for bullish).
        e.  Detect indicator extrema.
        f.  Identify divergences.
        g.  Store/Log found divergences.
6.  Handle errors gracefully at each step using `Result` and `anyhow`/`thiserror`.

**I. Output/Reporting**

*   Log found divergences to console (using `log` crate).
*   Store divergences in a dedicated PostgreSQL table.
    *   `CREATE TABLE divergences (id SERIAL PRIMARY KEY, symbol VARCHAR(20), timeframe VARCHAR(10), indicator_name VARCHAR(50), divergence_type VARCHAR(20), start_time TIMESTAMPTZ, end_time TIMESTAMPTZ, price_start REAL, price_end REAL, indicator_start REAL, indicator_end REAL, created_at TIMESTAMPTZ DEFAULT NOW());`
    *   Add a unique constraint on `(symbol, timeframe, indicator_name, divergence_type, start_time, end_time)` to prevent duplicates if re-running.
*   Optionally, write to CSV/JSON files.

---

**III. Robustness & Best Practices**

*   **Comprehensive Error Handling:** Use `Result<T, E>` extensively. Define custom error types with `thiserror` for domain-specific errors (e.g., `DataAlignmentError`, `InsufficientDataError`). Use `anyhow` for simpler error propagation in `main` or high-level functions.
*   **Logging:** Use `tracing` or `log` crate. Log key events, errors, and summary statistics.
*   **Testing:**
    *   **Unit Tests:** For peak/trough detection logic with mock data. For divergence rule implementations with mock extrema.
    *   **Integration Tests:** With a test PostgreSQL database. Test data fetching, alignment, and full divergence detection pipeline. Use `sqlx::test` macro.
*   **Concurrency:** Use `tokio` to process multiple symbol/timeframe pairs concurrently if I/O bound (DB queries).
    *   `futures::stream::StreamExt::for_each_concurrent` can be useful.
*   **Data Validation:** Check for empty datasets, consistency.
*   **Idempotency:** If storing results in DB, ensure re-running the finder doesn't create duplicate entries (e.g., using `ON CONFLICT DO NOTHING` or checking existence before insert).
*   **Code Organization:** Use modules (`mod`) effectively.
*   **Performance Profiling:** If performance becomes an issue, use tools like `perf` (Linux) or `Instruments` (macOS), and Rust's `criterion` for benchmarking critical sections.
*   **Database Migrations:** If schema changes, use a tool like `sqlx-cli migrate`.

---

**IV. Future Enhancements**

*   **More Sophisticated Peak/Trough Detection:**
    *   ZigZag-like logic.
    *   Filtering by volume confirmation.
    *   Consider using libraries that might offer this (though implementing it yourself is a good learning exercise).
*   **Divergence Scoring/Strength:** Some divergences are stronger than others (e.g., based on the angle of divergence, number of bars, volume confirmation).
*   **Real-time Processing:** Adapt to listen for new candles/indicator updates and process incrementally.
*   **Visualization:** Integrate with a charting library or output data for external visualization tools.
*   **Machine Learning:** Train a model to classify or predict based on identified divergences.
*   **Support for more indicator types without code changes:** More dynamic indicator configuration.
*   **Backtesting Framework Integration:** Feed identified divergences into a backtester.

---

**Getting Started Steps:**

1.  **Project Setup:** `cargo new divergence_finder --bin` and add dependencies.
2.  **PostgreSQL Setup:** Create your database and tables for candles and at least one indicator (e.g., RSI). Populate with some sample data.
3.  **Configuration:** Create a `config.toml` file. Implement `src/config.rs`.
4.  **Models:** Implement `src/models.rs`.
5.  **DB Layer:** Implement `src/db.rs` to fetch candles and one indicator. Test it.
6.  **Alignment:** Implement logic to merge candle and indicator data into `Vec<TimeSeriesPoint>`.
7.  **Extrema Detection:** Implement `src/analysis/extrema.rs`. Test with various patterns.
8.  **Divergence Logic:** Implement `src/analysis/divergence.rs` for one type (e.g., Regular Bullish) first. Test thoroughly. Then add others.
9.  **Main Orchestration:** Wire everything up in `src/main.rs`.
10. **Output:** Implement logging and DB storage for results.
11. **Refine & Test:** Add more tests, improve error handling, consider performance