//! Completed-session OHLCV rules shared by research and strategy execution.
use crate::DataSet;
use chrono::NaiveDate;
use serde::{Deserialize, Serialize};
use serde_json::{Value, json};
use std::collections::{BTreeMap, BTreeSet};

pub const CONTRACT: &str = "fidc_daily_ohlcv_pattern_v1";

pub fn catalog() -> Value {
    json!({"contract":CONTRACT,"templates":{
        "strength":{"label":"趋势强势","parameters":{"momentum_window":[25,5,120],"fast_window":[20,2,60],"slow_window":[60,20,252]},"stages":["selection","buy"],"method":"收盘价>短均线>长均线，按区间动量排序；不是当日金叉。"},
        "breakout":{"label":"前高突破","parameters":{"high_window":[60,5,252],"volume_window":[10,2,60],"volume_multiple":[1.3,1,10],"max_upper_shadow":[0.1,0,1]},"stages":["selection","buy"],"method":"收盘突破此前N日最高价，量达到此前M日均量倍数，上影比例受限；参考窗口不含当日。"},
        "volume_spike":{"label":"放量上涨","parameters":{"volume_window":[5,2,60],"volume_multiple":[3.0,1,10]},"stages":["selection","buy"],"method":"当日上涨且量达到此前N日最大量的指定倍数；不等同价格创新高。"},
        "shrink_breakout":{"label":"缩量突破","parameters":{"spike_lookback":[5,2,30],"volume_window":[5,2,60],"volume_multiple":[3.0,1,10],"shrink_ratio":[0.5,0.01,1]},"stages":["selection","buy"],"method":"此前观察窗有放量日，今日收盘超过该日最高价，成交量不超过其指定比例。"},
        "ma_below":{"label":"均线下方","parameters":{"ma_window":[20,2,252]},"stages":["sell"],"method":"完整收盘价低于含当日的N日均线；独立卖出条件。"},
        "volume_down":{"label":"放量下跌","parameters":{"volume_window":[5,2,60],"volume_multiple":[3.0,1,10]},"stages":["sell"],"method":"当日下跌且量达到此前N日最大量的指定倍数。"}
    },"data_frequency":"1d","execution_policies":["next_session_open"]})
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PatternSpec {
    pub template: String,
    #[serde(default)]
    pub parameters: BTreeMap<String, Value>,
}
impl PatternSpec {
    pub fn validate(mut self) -> Result<Self, String> {
        let catalog = catalog();
        let definition = catalog["templates"]
            .get(&self.template)
            .ok_or("未登记的量价模板")?;
        let parameters = definition["parameters"].as_object().unwrap();
        if self.parameters.keys().any(|k| !parameters.contains_key(k)) {
            return Err("模板包含未知参数".into());
        }
        for (key, bounds) in parameters {
            let value = self.parameters.get(key).unwrap_or(&bounds[0]);
            let number = value
                .as_f64()
                .filter(|v| v.is_finite())
                .ok_or_else(|| format!("{key}必须为有限数值"))?;
            if number < bounds[1].as_f64().unwrap() || number > bounds[2].as_f64().unwrap() {
                return Err(format!("{key}超出允许范围"));
            }
            if key.ends_with("window") || key == "spike_lookback" {
                if number.fract() != 0.0 {
                    return Err(format!("{key}必须是整数"));
                }
                self.parameters.insert(key.clone(), json!(number as usize));
            } else {
                self.parameters.insert(key.clone(), json!(number));
            }
        }
        if self.template == "strength" && self.n("fast_window") >= self.n("slow_window") {
            return Err("短均线必须小于长均线".into());
        }
        Ok(self)
    }
    pub fn n(&self, key: &str) -> usize {
        self.parameters[key].as_u64().unwrap() as usize
    }
    pub fn v(&self, key: &str) -> f64 {
        self.parameters[key].as_f64().unwrap()
    }
    pub fn history_len(&self) -> usize {
        match self.template.as_str() {
            "strength" => self.n("slow_window").max(self.n("momentum_window") + 1),
            "breakout" => self.n("high_window").max(self.n("volume_window")) + 1,
            "volume_spike" | "volume_down" => self.n("volume_window") + 1,
            "ma_below" => self.n("ma_window").max(2),
            "shrink_breakout" => self.n("spike_lookback") + self.n("volume_window") + 1,
            _ => unreachable!(),
        }
    }
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PatternBar {
    pub date: NaiveDate,
    pub open: Option<f64>,
    pub high: Option<f64>,
    pub low: Option<f64>,
    pub close: Option<f64>,
    pub volume: Option<f64>,
    pub adjustment_factor_backward1: Option<f64>,
    pub paused: Option<bool>,
    #[serde(default)]
    pub source_path: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PatternSeries {
    pub symbol: String,
    #[serde(default)]
    pub name: Option<String>,
    #[serde(default)]
    pub listed_at: Option<NaiveDate>,
    pub bars: Vec<PatternBar>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct PatternResult {
    pub symbol: String,
    pub name: Option<String>,
    pub matched: bool,
    pub score: Option<f64>,
    pub checks: Vec<Value>,
    pub values: Value,
    pub anchor: Value,
    pub exclusion: Option<Value>,
}

fn number(v: Option<f64>, symbol: &str, day: NaiveDate, field: &str) -> Result<f64, String> {
    v.filter(|v|v.is_finite()).ok_or_else(||format!("pattern_input_invalid: symbol={symbol}, date={day}, field={field}, reason=missing_or_nonfinite"))
}
fn check(checks: &mut Vec<Value>, label: &str, actual: f64, operator: &str, threshold: f64) {
    let passed = match operator {
        ">" => actual > threshold,
        "<" => actual < threshold,
        ">=" => actual >= threshold,
        "<=" => actual <= threshold,
        _ => false,
    };
    checks.push(json!({"label":label,"actual":actual,"operator":operator,"threshold":threshold,"passed":passed}));
}

fn mean(mut values: impl ExactSizeIterator<Item = f64>) -> Result<f64, String> {
    let count = values.len();
    let first = values.next().ok_or("pattern_mean_empty")?;
    // Center before summation so an unchanged decimal price stays exactly unchanged.
    let result = first
        + values
            .map(|value| (value - first) / count as f64)
            .sum::<f64>();
    if !result.is_finite() {
        return Err("pattern_mean_nonfinite".into());
    }
    Ok(result)
}

/// No calendar compression, fill-forward prices or numerical substitutes.
pub fn evaluate(
    spec: &PatternSpec,
    days: &[NaiveDate],
    series: &PatternSeries,
) -> Result<PatternResult, String> {
    if days.len() != spec.history_len() || days.windows(2).any(|w| w[0] >= w[1]) {
        return Err("pattern_calendar_incomplete: 需要完整、唯一且递增的真实交易日窗口".into());
    }
    let by_day = series
        .bars
        .iter()
        .map(|b| (b.date, b))
        .collect::<BTreeMap<_, _>>();
    if by_day.len() != series.bars.len() || series.bars.iter().any(|b| !days.contains(&b.date)) {
        return Err(format!(
            "pattern_input_invalid: symbol={}, reason=duplicate_or_out_of_scope",
            series.symbol
        ));
    }
    let mut unavailable = Vec::new();
    let mut prices = Vec::new();
    for &day in days {
        let Some(b) = by_day.get(&day) else {
            if series.listed_at.is_some_and(|listed| day < listed) {
                unavailable.push(
                    json!({"date":day,"reason":"before_listing","listed_at":series.listed_at}),
                );
                continue;
            }
            return Err(format!(
                "pattern_input_invalid: symbol={}, date={day}, reason=missing_market_row",
                series.symbol
            ));
        };
        let factor = if b.close.is_some_and(|c| c.is_finite() && c > 0.0) {
            let factor = number(
                b.adjustment_factor_backward1,
                &series.symbol,
                day,
                "adjustment_factor_backward1",
            )?;
            if factor <= 0.0 {
                return Err(format!(
                    "pattern_input_invalid: symbol={}, date={day}, field=adjustment_factor_backward1, reason=nonpositive",
                    series.symbol
                ));
            }
            factor
        } else {
            1.0
        };
        let paused = b.paused.ok_or_else(|| {
            format!(
                "pattern_input_invalid: symbol={}, date={day}, field=paused",
                series.symbol
            )
        })?;
        if paused {
            unavailable.push(
                json!({"date":day,"reason":"confirmed_suspension","source_path":b.source_path}),
            );
            continue;
        }
        if series.listed_at.is_some_and(|listed| day < listed) {
            return Err(format!(
                "pattern_input_invalid: symbol={}, date={day}, reason=price_before_listing",
                series.symbol
            ));
        }
        let o = number(b.open, &series.symbol, day, "open")?;
        let h = number(b.high, &series.symbol, day, "high")?;
        let l = number(b.low, &series.symbol, day, "low")?;
        let c = number(b.close, &series.symbol, day, "close")?;
        let v = number(b.volume, &series.symbol, day, "volume")?;
        if o <= 0.0 || l <= 0.0 || c <= 0.0 || h < o.max(c) || l > o.min(c) || v < 0.0 {
            return Err(format!(
                "pattern_input_invalid: symbol={}, date={day}, reason=invalid_ohlcv",
                series.symbol
            ));
        }
        prices.push((o * factor, h * factor, l * factor, c * factor, v));
    }
    let mut result = PatternResult {
        symbol: series.symbol.clone(),
        name: series.name.clone(),
        matched: false,
        score: None,
        checks: vec![],
        values: json!({}),
        anchor: Value::Null,
        exclusion: None,
    };
    if !unavailable.is_empty() {
        result.exclusion = Some(
            json!({"reason":"proven_incomplete_window","signal_date":days.last(),"evidence":unavailable}),
        );
        return Ok(result);
    }
    let len = prices.len();
    let (o, h, l, c, v) = prices[len - 1];
    let change = c / prices[len - 2].3 - 1.0;
    result.values = json!({"close":by_day[&days[len-1]].close,"daily_return":change});
    result.anchor = json!({"date":days[len-1],"raw_close":by_day[&days[len-1]].close,"factor":by_day[&days[len-1]].adjustment_factor_backward1});
    let mut score = None;
    match spec.template.as_str() {
        "strength" => {
            let fast = mean(prices[len - spec.n("fast_window")..].iter().map(|b| b.3))?;
            let slow = mean(prices[len - spec.n("slow_window")..].iter().map(|b| b.3))?;
            let momentum = c / prices[len - 1 - spec.n("momentum_window")].3 - 1.0;
            score = Some(momentum);
            result.values["momentum"] = json!(momentum);
            result.values["fast_ma"] = json!(fast);
            result.values["slow_ma"] = json!(slow);
            check(&mut result.checks, "收盘高于短均线", c, ">", fast);
            check(&mut result.checks, "短均线高于长均线", fast, ">", slow);
        }
        "breakout" => {
            let prior_high = prices[len - 1 - spec.n("high_window")..len - 1]
                .iter()
                .map(|b| b.1)
                .fold(f64::NEG_INFINITY, f64::max);
            let avg = mean(
                prices[len - 1 - spec.n("volume_window")..len - 1]
                    .iter()
                    .map(|b| b.4),
            )?;
            if avg <= 0.0 {
                return Err(format!(
                    "pattern_input_invalid: symbol={}, reason=zero_reference_volume",
                    series.symbol
                ));
            }
            let shadow = if h > l { (h - o.max(c)) / (h - l) } else { 0.0 };
            score = Some(c / prior_high - 1.0);
            result.values["volume_ratio"] = json!(v / avg);
            result.values["upper_shadow"] = json!(shadow);
            check(&mut result.checks, "收盘突破前高", c, ">", prior_high);
            check(
                &mut result.checks,
                "均量倍数",
                v / avg,
                ">=",
                spec.v("volume_multiple"),
            );
            check(
                &mut result.checks,
                "上影比例",
                shadow,
                "<=",
                spec.v("max_upper_shadow"),
            );
        }
        "volume_spike" | "volume_down" => {
            let high = prices[len - 1 - spec.n("volume_window")..len - 1]
                .iter()
                .map(|b| b.4)
                .fold(0.0, f64::max);
            if high <= 0.0 {
                return Err(format!(
                    "pattern_input_invalid: symbol={}, reason=zero_reference_volume",
                    series.symbol
                ));
            }
            score = Some(v / high);
            result.values["volume_ratio"] = json!(v / high);
            check(
                &mut result.checks,
                "最大量倍数",
                v / high,
                ">=",
                spec.v("volume_multiple"),
            );
            check(
                &mut result.checks,
                if spec.template == "volume_spike" {
                    "当日上涨"
                } else {
                    "当日下跌"
                },
                change,
                if spec.template == "volume_spike" {
                    ">"
                } else {
                    "<"
                },
                0.0,
            );
        }
        "ma_below" => {
            let avg = mean(prices[len - spec.n("ma_window")..].iter().map(|b| b.3))?;
            score = Some(avg / c - 1.0);
            result.values["ma"] = json!(avg);
            check(&mut result.checks, "收盘低于均线", c, "<", avg);
        }
        "shrink_breakout" => {
            let mut spikes = Vec::new();
            let mut eligible = Vec::new();
            for i in len - 1 - spec.n("spike_lookback")..len - 1 {
                let prior = prices[i - spec.n("volume_window")..i]
                    .iter()
                    .map(|b| b.4)
                    .fold(0.0, f64::max);
                if prior <= 0.0 {
                    return Err(format!(
                        "pattern_input_invalid: symbol={}, date={}, reason=zero_reference_volume",
                        series.symbol, days[i]
                    ));
                }
                if prices[i].4 >= prior * spec.v("volume_multiple") {
                    spikes.push(i);
                    if c > prices[i].1 && v <= prices[i].4 * spec.v("shrink_ratio") {
                        eligible.push(i);
                    }
                }
            }
            check(
                &mut result.checks,
                "观察窗存在放量日",
                spikes.len() as f64,
                ">",
                0.0,
            );
            if let Some(&i) = eligible.last().or_else(|| spikes.last()) {
                score = Some(c / prices[i].1 - 1.0);
                result.values["spike_date"] = json!(days[i]);
                result.values["volume_ratio"] = json!(v / prices[i].4);
                check(
                    &mut result.checks,
                    "收盘突破放量日高点",
                    c,
                    ">",
                    prices[i].1,
                );
                check(
                    &mut result.checks,
                    "缩量比例",
                    v / prices[i].4,
                    "<=",
                    spec.v("shrink_ratio"),
                );
            }
        }
        _ => unreachable!(),
    }
    if score.is_some_and(|v| !v.is_finite()) {
        return Err("pattern_result_nonfinite".into());
    }
    result.score = score;
    result.matched = result.checks.iter().all(|c| c["passed"] == true);
    Ok(result)
}

pub fn evaluate_dataset(
    spec: &PatternSpec,
    data: &DataSet,
    date: NaiveDate,
    symbol: &str,
) -> Result<PatternResult, String> {
    let days = data.calendar().trailing_days(date, spec.history_len());
    let bars = days
        .iter()
        .filter_map(|&d| {
            data.market(d, symbol).map(|b| PatternBar {
                date: d,
                open: Some(b.open),
                high: Some(b.high),
                low: Some(b.low),
                close: Some(b.close),
                volume: Some(b.volume as f64),
                adjustment_factor_backward1: data
                    .factor(d, symbol)
                    .and_then(|f| f.adjustment_factor_backward1),
                paused: Some(b.paused),
                source_path: None,
            })
        })
        .collect();
    evaluate(
        spec,
        &days,
        &PatternSeries {
            symbol: symbol.into(),
            name: None,
            listed_at: data.instrument(symbol).and_then(|i| i.listed_at),
            bars,
        },
    )
}

pub fn evaluate_batch(
    spec: PatternSpec,
    days: &[NaiveDate],
    series: &[PatternSeries],
) -> Result<Value, String> {
    let spec = spec.validate()?;
    if series.is_empty()
        || series.len() > 200
        || series
            .iter()
            .map(|s| &s.symbol)
            .collect::<BTreeSet<_>>()
            .len()
            != series.len()
    {
        return Err("pattern_batch_invalid: 需要1至200只唯一证券".into());
    }
    let rows = series
        .iter()
        .map(|s| evaluate(&spec, days, s))
        .collect::<Result<Vec<_>, _>>()?;
    Ok(
        json!({"contract":CONTRACT,"spec":spec,"required_history":spec.history_len(),"rows":rows,"read_only":true}),
    )
}

pub fn expression_specs(expression: &str) -> Result<Vec<PatternSpec>, String> {
    let mut specs = Vec::new();
    for helper in ["pattern_signal", "pattern_score"] {
        for (index, _) in expression.match_indices(helper) {
            if index > 0
                && expression[..index]
                    .chars()
                    .next_back()
                    .is_some_and(|c| c.is_alphanumeric() || c == '_')
            {
                continue;
            }
            let rest = expression[index + helper.len()..].trim_start();
            let Some(rest) = rest.strip_prefix('(') else {
                continue;
            };
            let rest = rest.trim_start();
            let mut stream = serde_json::Deserializer::from_str(rest).into_iter::<String>();
            let text = stream
                .next()
                .ok_or("missing pattern JSON")?
                .map_err(|e| e.to_string())?;
            if !rest[stream.byte_offset()..].trim_start().starts_with(')') {
                return Err("pattern helper takes one JSON string".into());
            }
            let spec: PatternSpec = serde_json::from_str(&text).map_err(|e| e.to_string())?;
            specs.push(spec.validate()?);
        }
    }
    Ok(specs)
}

#[cfg(test)]
mod tests {
    use super::*;
    fn fixture(template: &str) -> (PatternSpec, Vec<NaiveDate>, PatternSeries) {
        let spec = PatternSpec {
            template: template.into(),
            parameters: BTreeMap::new(),
        }
        .validate()
        .unwrap();
        let days = (0..spec.history_len())
            .map(|n| {
                NaiveDate::from_ymd_opt(2025, 1, 1).unwrap() + chrono::Duration::days(n as i64)
            })
            .collect::<Vec<_>>();
        let bars = days
            .iter()
            .enumerate()
            .map(|(n, &date)| {
                let c = 10.0 + n as f64;
                PatternBar {
                    date,
                    open: Some(c),
                    high: Some(c),
                    low: Some(c),
                    close: Some(c),
                    volume: Some(1000.0),
                    adjustment_factor_backward1: Some(1.0),
                    paused: Some(false),
                    source_path: Some("fixture.parquet".into()),
                }
            })
            .collect();
        (
            spec,
            days,
            PatternSeries {
                symbol: "000001.SZ".into(),
                name: None,
                listed_at: Some(NaiveDate::from_ymd_opt(1991, 4, 3).unwrap()),
                bars,
            },
        )
    }
    #[test]
    fn daily_patterns_all_templates_and_score_absence() {
        for template in [
            "strength",
            "breakout",
            "volume_spike",
            "shrink_breakout",
            "ma_below",
            "volume_down",
        ] {
            let (spec, days, series) = fixture(template);
            let result = evaluate(&spec, &days, &series).unwrap();
            assert_eq!(result.matched, template == "strength");
            assert_eq!(result.score.is_none(), template == "shrink_breakout");
        }
    }
    #[test]
    fn daily_patterns_adjusts_all_prices_not_volume() {
        let (spec, days, series) = fixture("strength");
        let a = evaluate(&spec, &days, &series).unwrap();
        let mut split = series.clone();
        for b in &mut split.bars {
            b.open = b.open.map(|p| p / 2.0);
            b.high = b.high.map(|p| p / 2.0);
            b.low = b.low.map(|p| p / 2.0);
            b.close = b.close.map(|p| p / 2.0);
            b.adjustment_factor_backward1 = Some(2.0);
        }
        let b = evaluate(&spec, &days, &split).unwrap();
        assert_eq!(a.score, b.score);
        assert_eq!(a.checks, b.checks);
    }

    #[test]
    fn daily_patterns_flat_decimal_prices_do_not_create_a_sell_signal() {
        let (mut spec, _, mut series) = fixture("strength");
        spec.template = "ma_below".into();
        spec.parameters = BTreeMap::from([("ma_window".into(), json!(60))]);
        for bar in &mut series.bars {
            bar.open = Some(10.1);
            bar.high = Some(10.1);
            bar.low = Some(10.1);
            bar.close = Some(10.1);
        }
        let days = series.bars.iter().map(|bar| bar.date).collect::<Vec<_>>();
        let result = evaluate(&spec, &days, &series).unwrap();
        assert!(
            !result.matched,
            "unchanged decimal prices must not trigger a below-MA sell: {:?}",
            result.checks
        );
        assert_eq!(result.values["ma"], 10.1);
    }
    #[test]
    fn daily_patterns_no_missing_data_fallback() {
        let (spec, days, mut series) = fixture("strength");
        series.bars[0].adjustment_factor_backward1 = None;
        assert!(
            evaluate(&spec, &days, &series)
                .unwrap_err()
                .contains("adjustment_factor")
        );
        series.bars[0].paused = Some(true);
        assert!(evaluate(&spec, &days, &series).is_err());
        series.bars[0].adjustment_factor_backward1 = Some(1.0);
        let excluded = evaluate(&spec, &days, &series).unwrap();
        assert!(excluded.exclusion.is_some());
        assert!(!excluded.matched);
        series.bars.remove(0);
        assert!(evaluate(&spec, &days, &series).is_err());
        series.listed_at = Some(days[1]);
        assert!(evaluate(&spec, &days, &series).unwrap().exclusion.is_some());
    }
    #[test]
    fn daily_patterns_rejects_future_and_duplicate_bars() {
        let (spec, days, mut series) = fixture("strength");
        series.bars.push(series.bars[0].clone());
        assert!(evaluate(&spec, &days, &series).is_err());
        series.bars.last_mut().unwrap().date = *days.last().unwrap() + chrono::Duration::days(1);
        assert!(evaluate(&spec, &days, &series).is_err());
    }
    #[test]
    fn daily_patterns_helper_literal_preserves_parameters() {
        let text =
            serde_json::to_string(&json!({"template":"breakout","parameters":{"high_window":252}}))
                .unwrap();
        let expression = format!("pattern_signal({})", serde_json::to_string(&text).unwrap());
        assert_eq!(expression_specs(&expression).unwrap()[0].history_len(), 253);
        assert!(expression_specs("pattern_signal(\"{}\")").is_err());
    }
}