mirror of
https://github.com/Sabermrddz/QuantCore-FX.git
synced 2026-07-28 02:57:51 +00:00
saber
338 lines
9.9 KiB
Python
338 lines
9.9 KiB
Python
"""
|
|
APEX Layer 1 — Scoring Engine
|
|
|
|
Implements the exact scoring formula from the README:
|
|
1. Collect raw values (rates, CPI deviations, PMI)
|
|
2. Calculate derived values
|
|
3. Normalize each input to 0-100 using min-max scaling
|
|
4. Apply weights (Rate 50%, CPI 30%, PMI 20%)
|
|
5. Sum to get final score
|
|
6. Rank currencies and pair strongest vs weakest
|
|
7. Validate gap >= MIN_GAP to trade
|
|
|
|
All scores are 0-100. Gap must be >= 20 to generate a trade signal.
|
|
"""
|
|
|
|
from typing import Dict, List, Tuple, Optional
|
|
import config
|
|
|
|
|
|
def normalise(values: List[float]) -> List[float]:
|
|
"""
|
|
Normalize a list of values to 0-100 using min-max scaling.
|
|
|
|
If all values are equal (no variation), return [50.0] * len(values)
|
|
to represent perfect neutrality.
|
|
|
|
Args:
|
|
values: List of numeric values
|
|
|
|
Returns:
|
|
List of normalized values (0.0 to 100.0)
|
|
"""
|
|
if not values:
|
|
return []
|
|
|
|
min_v = min(values)
|
|
max_v = max(values)
|
|
|
|
# Handle edge case: all values identical
|
|
if max_v == min_v:
|
|
return [50.0] * len(values)
|
|
|
|
# Min-max scaling to [0, 100]
|
|
return [(v - min_v) / (max_v - min_v) * 100.0 for v in values]
|
|
|
|
|
|
def calculate_rate_differentials(rates: Dict[str, Optional[float]]) -> Dict[str, Optional[float]]:
|
|
"""
|
|
Calculate interest rate differential for each currency vs G8 average.
|
|
|
|
rate_diff[i] = rate[i] - mean(rate)
|
|
|
|
Args:
|
|
rates: Dict mapping currency to interest rate (% or None)
|
|
|
|
Returns:
|
|
Dict mapping currency to rate differential
|
|
"""
|
|
# Filter out None values for average calculation
|
|
valid_rates = [r for r in rates.values() if r is not None]
|
|
|
|
if not valid_rates:
|
|
# All rates missing — return zeros
|
|
return {currency: 0.0 for currency in config.CURRENCIES}
|
|
|
|
avg_rate = sum(valid_rates) / len(valid_rates)
|
|
|
|
# Calculate differentials (None becomes 0.0)
|
|
differentials = {}
|
|
for currency in config.CURRENCIES:
|
|
rate = rates.get(currency)
|
|
differentials[currency] = (rate - avg_rate) if rate is not None else 0.0
|
|
|
|
return differentials
|
|
|
|
|
|
def calculate_cpi_deviations(cpi_values: Dict[str, Optional[float]]) -> Dict[str, Optional[float]]:
|
|
"""
|
|
Calculate CPI deviation for each currency vs its CB target.
|
|
|
|
cpi_dev[i] = actual_cpi[i] - target[i]
|
|
|
|
- Positive deviation (above target) → hawkish pressure (stronger score)
|
|
- Negative deviation (below target) → dovish pressure (weaker score)
|
|
|
|
Args:
|
|
cpi_values: Dict mapping currency to actual CPI % (or None)
|
|
|
|
Returns:
|
|
Dict mapping currency to CPI deviation
|
|
"""
|
|
deviations = {}
|
|
for currency in config.CURRENCIES:
|
|
cpi = cpi_values.get(currency)
|
|
target = config.CB_TARGETS.get(currency, 2.0)
|
|
|
|
# None becomes 0.0 deviation (neutral)
|
|
deviations[currency] = (cpi - target) if cpi is not None else 0.0
|
|
|
|
return deviations
|
|
|
|
|
|
def score_all_currencies(
|
|
rates: Dict[str, Optional[float]],
|
|
cpi_values: Dict[str, Optional[float]],
|
|
pmi_values: Dict[str, Optional[float]]
|
|
) -> Dict[str, Dict]:
|
|
"""
|
|
Calculate the complete score for all 8 currencies.
|
|
|
|
Steps:
|
|
1. Calculate rate differentials
|
|
2. Calculate CPI deviations
|
|
3. Normalize each input to 0-100
|
|
4. Apply weights
|
|
5. Sum to get total score
|
|
6. Rank by score
|
|
|
|
Args:
|
|
rates: Dict currency -> interest rate % (or None)
|
|
cpi_values: Dict currency -> actual CPI % (or None)
|
|
pmi_values: Dict currency -> PMI reading (or None)
|
|
|
|
Returns:
|
|
Dict mapping currency to:
|
|
{
|
|
'score_rate': float (0-100),
|
|
'score_cpi': float (0-100),
|
|
'score_pmi': float (0-100),
|
|
'total_score': float (0-100),
|
|
'rank': int (1-8)
|
|
}
|
|
"""
|
|
# Step 1 & 2: Calculate derived values
|
|
rate_diffs = calculate_rate_differentials(rates)
|
|
cpi_devs = calculate_cpi_deviations(cpi_values)
|
|
pmi_raws = pmi_values.copy() # Use PMI values as-is
|
|
|
|
# Extract numeric lists for normalization (skip None values)
|
|
rate_diff_list = [rate_diffs[c] for c in config.CURRENCIES]
|
|
cpi_dev_list = [cpi_devs[c] for c in config.CURRENCIES]
|
|
|
|
# For PMI, treat None as 50 (neutral) for normalization purposes
|
|
pmi_list = [pmi_raws.get(c) if pmi_raws.get(c) is not None else 50.0 for c in config.CURRENCIES]
|
|
|
|
# Step 3: Normalize each input to 0-100
|
|
norm_rate = normalise(rate_diff_list)
|
|
norm_cpi = normalise(cpi_dev_list)
|
|
norm_pmi = normalise(pmi_list)
|
|
|
|
# Step 4 & 5: Apply weights and calculate total scores
|
|
scores_raw = {}
|
|
for i, currency in enumerate(config.CURRENCIES):
|
|
total = (
|
|
norm_rate[i] * config.WEIGHT_RATE +
|
|
norm_cpi[i] * config.WEIGHT_CPI +
|
|
norm_pmi[i] * config.WEIGHT_PMI
|
|
)
|
|
|
|
scores_raw[currency] = {
|
|
'score_rate': norm_rate[i],
|
|
'score_cpi': norm_cpi[i],
|
|
'score_pmi': norm_pmi[i],
|
|
'total_score': total
|
|
}
|
|
|
|
# Step 6: Rank by total score (descending)
|
|
sorted_currencies = sorted(
|
|
scores_raw.items(),
|
|
key=lambda x: x[1]['total_score'],
|
|
reverse=True
|
|
)
|
|
|
|
# Add rank to each score
|
|
final_scores = {}
|
|
for rank, (currency, score_data) in enumerate(sorted_currencies, start=1):
|
|
score_data['rank'] = rank
|
|
final_scores[currency] = score_data
|
|
|
|
return final_scores
|
|
|
|
|
|
def get_ranked_list(scores: Dict[str, Dict]) -> List[Tuple[str, float, int]]:
|
|
"""
|
|
Get currencies sorted by score (highest first).
|
|
|
|
Args:
|
|
scores: Dict from score_all_currencies()
|
|
|
|
Returns:
|
|
List of (currency, total_score, rank) tuples
|
|
"""
|
|
return sorted(
|
|
[(c, s['total_score'], s['rank']) for c, s in scores.items()],
|
|
key=lambda x: x[1],
|
|
reverse=True
|
|
)
|
|
|
|
|
|
def pair_currencies(scores: Dict[str, Dict]) -> Tuple[str, str, float]:
|
|
"""
|
|
Get the strongest and weakest currencies (for pairing).
|
|
|
|
Returns:
|
|
Tuple of (strongest_currency, weakest_currency, gap)
|
|
"""
|
|
ranked = get_ranked_list(scores)
|
|
|
|
if not ranked or len(ranked) < 2:
|
|
raise ValueError("Cannot pair: insufficient scored currencies")
|
|
|
|
strongest_currency, strongest_score, _ = ranked[0]
|
|
weakest_currency, weakest_score, _ = ranked[-1]
|
|
gap = strongest_score - weakest_score
|
|
|
|
return strongest_currency, weakest_currency, gap
|
|
|
|
|
|
def generate_signal(scores: Dict[str, Dict]) -> Tuple[str, str, str]:
|
|
"""
|
|
Generate the primary trade signal.
|
|
|
|
Returns:
|
|
Tuple of (signal_text, status, gap_description)
|
|
|
|
signal_text: "SHORT {weakest}/{strongest}" or "NO TRADE"
|
|
status: "ACTIVE" or "NO_TRADE"
|
|
gap_description: e.g. "Gap: 74 points · Strong signal"
|
|
"""
|
|
strongest, weakest, gap = pair_currencies(scores)
|
|
|
|
if gap >= config.MIN_GAP_TO_TRADE:
|
|
signal_text = f"SHORT {weakest}/{strongest}"
|
|
status = "ACTIVE"
|
|
|
|
# Classify gap tier
|
|
if gap >= 60:
|
|
tier = "Strong signal"
|
|
elif gap >= config.GAP_THRESHOLDS["standard"]:
|
|
tier = "Standard signal"
|
|
elif gap >= config.GAP_THRESHOLDS["weak"]:
|
|
tier = "Weak signal"
|
|
else:
|
|
tier = "Marginal signal"
|
|
|
|
gap_desc = f"Gap: {gap:.1f} points · {tier}"
|
|
else:
|
|
signal_text = "NO TRADE"
|
|
status = "NO_TRADE"
|
|
gap_desc = f"Gap: {gap:.1f} points · Too narrow (< {config.MIN_GAP_TO_TRADE})"
|
|
|
|
return signal_text, status, gap_desc
|
|
|
|
|
|
def build_directional_bias_matrix(scores: Dict[str, Dict]) -> Dict[str, Dict]:
|
|
"""Build an advisory directional bias matrix from Layer 1 scores.
|
|
|
|
DISPLAY ONLY — does not gate or block any trade signal anywhere in the system.
|
|
Top 2 → "STRONG", Bottom 2 → "WEAK", Middle 4 → "NEUTRAL".
|
|
|
|
Returns:
|
|
{
|
|
"USD": {"direction": "STRONG", "score": 85.2, "rank": 1},
|
|
"JPY": {"direction": "WEAK", "score": 22.1, "rank": 8},
|
|
...
|
|
}
|
|
"""
|
|
ranked = get_ranked_list(scores)
|
|
n = len(ranked)
|
|
matrix = {}
|
|
for i, (currency, total_score, rank) in enumerate(ranked):
|
|
if i < 2 and n >= 4:
|
|
direction = "STRONG"
|
|
elif i >= n - 2 and n >= 4:
|
|
direction = "WEAK"
|
|
else:
|
|
direction = "NEUTRAL"
|
|
matrix[currency] = {
|
|
"direction": direction,
|
|
"score": total_score,
|
|
"rank": rank,
|
|
}
|
|
return matrix
|
|
|
|
|
|
def get_gap_tier(gap: float) -> str:
|
|
"""
|
|
Classify a gap size into trading tiers.
|
|
|
|
Returns:
|
|
One of: "no_trade", "weak", "standard", "strong"
|
|
"""
|
|
if gap < config.GAP_THRESHOLDS["weak"]:
|
|
return "no_trade"
|
|
elif gap < config.GAP_THRESHOLDS["standard"]:
|
|
return "weak"
|
|
elif gap < 60:
|
|
return "standard"
|
|
else:
|
|
return "strong"
|
|
|
|
|
|
def validate_scores(scores: Dict[str, Dict]) -> bool:
|
|
"""
|
|
Validate that scores dict has all required fields.
|
|
|
|
Args:
|
|
scores: Dict from score_all_currencies()
|
|
|
|
Returns:
|
|
True if valid, raises ValueError if invalid
|
|
"""
|
|
required_fields = {'score_rate', 'score_cpi', 'score_pmi', 'total_score', 'rank'}
|
|
|
|
for currency in config.CURRENCIES:
|
|
if currency not in scores:
|
|
raise ValueError(f"Missing scores for {currency}")
|
|
|
|
score_data = scores[currency]
|
|
missing = required_fields - set(score_data.keys())
|
|
|
|
if missing:
|
|
raise ValueError(
|
|
f"Missing fields for {currency}: {missing}"
|
|
)
|
|
|
|
# Check value ranges
|
|
for field in ['score_rate', 'score_cpi', 'score_pmi', 'total_score']:
|
|
value = score_data[field]
|
|
if not (0 <= value <= 100):
|
|
raise ValueError(
|
|
f"{currency}.{field} out of range [0-100]: {value}"
|
|
)
|
|
|
|
return True
|
|
|
|
|