| 1 | """Deterministic recommendation policy, independent of membership and public scoring. |
| 2 | |
| 3 | V1 thresholds are deliberately explicit initial policy, not calibrated forecasts. |
| 4 | All prices use the persisted technical snapshot's currency and unadjusted basis. |
| 5 | """ |
| 6 | from copy import deepcopy |
| 7 | from hashlib import sha256 |
| 8 | import json |
| 9 | from math import isfinite |
| 10 | |
| 11 | VERSION = 'RECOMMENDATION_ENGINE_V1' |
| 12 | BUY_SCORE, STRONG_SCORE, MIN_CONFIDENCE, MIN_COVERAGE = 65, 80, 60, 60 |
| 13 | RANGE_KEYS = ('short_entry_low short_entry_high short_target_1 short_target_2 short_invalidation ' |
| 14 | 'long_entry_low long_entry_high long_fair_value long_target long_invalidation').split() |
| 15 | AREA_NAMES = {'QUALITY': 'FUNDAMENTAL_BUSINESS_QUALITY', 'QUARTERLY': 'QUARTERLY_EARNINGS_TREND', |
| 16 | 'CATALYST': 'ORDER_BOOK_CAPACITY_CATALYSTS', 'NEWS': 'NEWS_GEOPOLITICAL_EVENTS', |
| 17 | 'GOVERNANCE': 'MANAGEMENT_GOVERNANCE'} |
| 18 | |
| 19 | |
| 20 | def area_scores(rule): |
| 21 | raw = {a['area']: a.get('raw_score') for a in rule.get('area_scores', [])} |
| 22 | return {**raw, **{k: raw.get(v) for k, v in AREA_NAMES.items()}} |
| 23 | |
| 24 | |
| 25 | def number(value): |
| 26 | return float(value) if isinstance(value, (int, float)) and not isinstance(value, bool) and isfinite(value) else None |
| 27 | |
| 28 | |
| 29 | def digest(value): |
| 30 | return sha256(json.dumps(value, sort_keys=True, separators=(',', ':'), allow_nan=False).encode()).hexdigest() |
| 31 | |
| 32 | |
| 33 | def ranges(snapshot): |
| 34 | result = dict.fromkeys(RANGE_KEYS) |
| 35 | technical = snapshot['evidence_state'].get('technical', {}) |
| 36 | p, support, resistance, atr = (number(v) for v in ( |
| 37 | snapshot.get('current_price'), technical.get('support_level'), |
| 38 | technical.get('resistance_level'), technical.get('atr14'))) |
| 39 | if p and support and resistance and 0 < support <= p < resistance: |
| 40 | result.update(short_entry_low=support, short_entry_high=min(p, support * 1.03), |
| 41 | short_target_1=resistance) |
| 42 | if atr and atr > 0 and support > atr: |
| 43 | result.update(short_target_2=resistance + atr, short_invalidation=support - atr) |
| 44 | # Only explicitly supplied fair-value evidence is usable; never turn a valuation score into a price. |
| 45 | valuation = snapshot['evidence_state'].get('valuation', {}) |
| 46 | fair, bull, invalidation = (number(valuation.get(k)) for k in ('fair_value', 'bull_target', 'invalidation')) |
| 47 | if fair and fair > 0: |
| 48 | result.update(long_entry_low=fair * .8, long_entry_high=fair * .9, long_fair_value=fair) |
| 49 | if bull and bull >= fair: |
| 50 | result['long_target'] = bull |
| 51 | if invalidation and 0 < invalidation < fair * .8: |
| 52 | result['long_invalidation'] = invalidation |
| 53 | return {k: round(v, 4) if v is not None else None for k, v in result.items()} |
| 54 | |
| 55 | |
| 56 | class RecommendationEngineV1: |
| 57 | def evaluate(self, snapshot, previous=None): |
| 58 | evidence = snapshot['evidence_state'] |
| 59 | rule = evidence.get('rule', {}) |
| 60 | dimensions = area_scores(rule) |
| 61 | critical = any(r.get('severity') == 'CRITICAL' for r in rule.get('risk_overrides', [])) |
| 62 | weak = [a for a in ('QUALITY', 'GROWTH', 'BALANCE_SHEET', 'QUARTERLY') |
| 63 | if dimensions.get(a) is not None and dimensions[a] <= 30] |
| 64 | governance = dimensions.get('GOVERNANCE') |
| 65 | thesis_broken = critical or (governance is not None and governance <= 20) or len(weak) >= 2 |
| 66 | score = snapshot.get('opportunity_score') |
| 67 | qualified = (snapshot['rank_eligible'] and score is not None and score >= BUY_SCORE |
| 68 | and snapshot['opportunity_confidence'] >= MIN_CONFIDENCE |
| 69 | and snapshot['score_coverage'] >= MIN_COVERAGE and not thesis_broken) |
| 70 | technical = evidence.get('technical', {}).get('technical_state') |
| 71 | short_buy = qualified and technical in {'UPTREND', 'BREAKOUT', 'PULLBACK_IN_UPTREND', 'REVERSAL_CANDIDATE'} |
| 72 | long_buy = qualified and all(dimensions.get(a) is not None and dimensions[a] >= 60 |
| 73 | for a in ('QUALITY', 'VALUATION')) |
| 74 | result = {k: deepcopy(snapshot.get(k)) for k in ( |
| 75 | 'global_instrument_id', 'market', 'symbol', 'company_name', 'generated_at', 'rule_engine_version', 'ranker_version', |
| 76 | 'opportunity_score', 'top_positive_reasons', 'top_negative_reasons')} |
| 77 | result.update(recommendation_engine_version=VERSION, price_at_recommendation=snapshot.get('current_price'), |
| 78 | confidence=snapshot['opportunity_confidence'], coverage=snapshot['score_coverage'], |
| 79 | new_investor_action='AVOID' if thesis_broken else ('STRONG_BUY_CANDIDATE' if score >= STRONG_SCORE else 'BUY_CANDIDATE') if qualified else 'WATCH_WAIT', |
| 80 | existing_holder_action='EXIT_REVIEW' if thesis_broken else 'TOP_UP' if long_buy else 'HOLD_NO_NEW_MONEY' if not qualified else 'HOLD', |
| 81 | short_term_action='EXIT' if thesis_broken else 'BUY' if short_buy else 'WAIT', |
| 82 | long_term_action='EXIT_REVIEW' if thesis_broken else 'ACCUMULATE' if long_buy else 'REDUCE' if weak else 'HOLD', |
| 83 | evidence_snapshot=deepcopy(evidence), **ranges(snapshot)) |
| 84 | reasons = result['top_negative_reasons'] or [] |
| 85 | if any(result[k] is None for k in RANGE_KEYS): |
| 86 | reasons = [*reasons, 'PRICE_RANGE_EVIDENCE_INSUFFICIENT'] |
| 87 | if thesis_broken: |
| 88 | reasons = [*reasons, 'LONG_TERM_THESIS_BROKEN'] |
| 89 | result['top_negative_reasons'] = sorted(set(reasons)) |
| 90 | if previous: |
| 91 | comparison = lifecycle(previous, result, snapshot.get('current_price'), {'review': True}) |
| 92 | result['short_term_action'] = comparison['current_short_action'] |
| 93 | result['long_term_action'] = comparison['current_long_action'] |
| 94 | if comparison['current_long_action'] == 'EXIT_REVIEW': |
| 95 | result['existing_holder_action'] = 'EXIT_REVIEW' |
| 96 | elif comparison['current_long_action'] == 'REDUCE' or comparison['current_short_action'] == 'EXIT': |
| 97 | result['existing_holder_action'] = 'REDUCE' |
| 98 | elif comparison['current_short_action'] == 'PARTIAL_PROFIT': |
| 99 | result['existing_holder_action'] = 'PARTIAL_PROFIT' |
| 100 | if result['long_term_action'] != 'TOP_UP': |
| 101 | result['new_investor_action'] = 'WATCH_WAIT' |
| 102 | result['top_negative_reasons'] = sorted(set(result['top_negative_reasons'] + comparison['lifecycle_reasons'])) |
| 103 | # Cache clocks, completion timestamps and observed price drift are not evidence changes. |
| 104 | evidence_key = {'references': rule.get('evidence_references', []), 'dimensions': dimensions, |
| 105 | 'metrics': [a.get('metrics', []) for a in rule.get('area_scores', [])], |
| 106 | 'risk': rule.get('risk_overrides', []), 'technical_state': technical, |
| 107 | 'sector_state': evidence.get('sector', {}).get('sector_state'), |
| 108 | 'missing': evidence.get('missing_inputs', []), 'stale': evidence.get('stale_inputs', [])} |
| 109 | p = result['price_at_recommendation'] |
| 110 | result['fingerprint'] = digest({ |
| 111 | 'instrument': result['global_instrument_id'], 'engine': VERSION, |
| 112 | 'actions': [result[k] for k in ('new_investor_action', 'existing_holder_action', 'short_term_action', 'long_term_action')], |
| 113 | 'reference_bucket': round(p, -1) if p is not None else None, |
| 114 | 'ranges': {k: result[k] for k in RANGE_KEYS}, 'evidence': evidence_key, |
| 115 | 'score_state': [round(result[k] / 5) * 5 if result[k] is not None else None for k in ('opportunity_score', 'confidence', 'coverage')]}) |
| 116 | return result |
| 117 | |
| 118 | |
| 119 | def lifecycle(anchor, recommendation, price, previous_state=None): |
| 120 | """Compare with the original trade levels. Never overwrite the historical anchor.""" |
| 121 | reasons = [] |
| 122 | short, long = recommendation['short_term_action'], recommendation['long_term_action'] |
| 123 | ss = ls = 'NEW' |
| 124 | p = number(price) |
| 125 | for horizon in ('short', 'long'): |
| 126 | invalidation = anchor.get(f'{horizon}_invalidation') |
| 127 | low, high = anchor.get(f'{horizon}_entry_low'), anchor.get(f'{horizon}_entry_high') |
| 128 | target = anchor.get('short_target_1' if horizon == 'short' else 'long_target') |
| 129 | state = 'HOLDING' if previous_state else 'NEW' |
| 130 | if p is not None: |
| 131 | if invalidation and p <= invalidation: |
| 132 | state = 'INVALIDATED' |
| 133 | elif invalidation and p <= invalidation * 1.03: |
| 134 | state = 'INVALIDATION_APPROACHING' |
| 135 | elif target and p >= target: |
| 136 | state = 'TARGET_REACHED' |
| 137 | elif target and p >= target * .97: |
| 138 | state = 'TARGET_APPROACHING' |
| 139 | elif low is not None and high is not None and low <= p <= high: |
| 140 | state = 'IN_ENTRY_ZONE' |
| 141 | elif high and high < p <= high * 1.03: |
| 142 | state = 'ENTRY_APPROACHING' |
| 143 | if horizon == 'short': |
| 144 | ss = state |
| 145 | if state == 'INVALIDATED': |
| 146 | short = 'EXIT' |
| 147 | reasons.append('SHORT_INVALIDATION_REACHED') |
| 148 | elif state == 'TARGET_REACHED': |
| 149 | ss, short = 'PARTIAL_PROFIT', 'PARTIAL_PROFIT' |
| 150 | reasons.extend(['TARGET_1_REACHED', 'SHORT_TERM_RISK_REWARD_COMPRESSED']) |
| 151 | if anchor.get('short_target_2') and p >= anchor['short_target_2'] * .97: |
| 152 | reasons.append('TARGET_2_REACHED' if p >= anchor['short_target_2'] else 'TARGET_2_APPROACHING') |
| 153 | elif (previous_state and short == 'WAIT' |
| 154 | and anchor.get('short_term_action') in {'BUY', 'HOLD', 'PARTIAL_PROFIT'}): |
| 155 | short = 'HOLD' |
| 156 | else: |
| 157 | ls = state |
| 158 | if state == 'INVALIDATED': |
| 159 | long, ls = 'EXIT_REVIEW', 'INVALIDATED' |
| 160 | elif state in {'TARGET_REACHED', 'TARGET_APPROACHING'} and long not in {'REDUCE', 'EXIT_REVIEW'}: |
| 161 | long = 'HOLD' |
| 162 | elif (previous_state and long == 'ACCUMULATE' and p is not None |
| 163 | and anchor.get('price_at_recommendation') and p <= anchor['price_at_recommendation'] * .97): |
| 164 | long = 'TOP_UP' |
| 165 | if recommendation['long_term_action'] == 'EXIT_REVIEW': |
| 166 | long, ls = 'EXIT_REVIEW', 'EXIT_REVIEW' |
| 167 | reasons.append('LONG_TERM_THESIS_BROKEN') |
| 168 | elif long == 'REDUCE': |
| 169 | ls = 'THESIS_WEAKENING' |
| 170 | if short == 'PARTIAL_PROFIT' and long in {'TOP_UP', 'ACCUMULATE'}: |
| 171 | long = 'HOLD' |
| 172 | if recommendation['short_term_action'] == 'EXIT': |
| 173 | short, ss = 'EXIT', 'EXIT_REVIEW' |
| 174 | def distance(key): |
| 175 | return round((anchor[key] / p - 1) * 100, 4) if p and anchor.get(key) else None |
| 176 | return dict(short_term_state=ss, long_term_state=ls, current_short_action=short, |
| 177 | current_long_action=long, lifecycle_status=ls if ls in {'EXIT_REVIEW', 'INVALIDATED', 'THESIS_WEAKENING'} else ss, |
| 178 | price_at_recommendation=anchor.get('price_at_recommendation'), current_price=p, |
| 179 | short_target_distance_pct=distance('short_target_2'), long_target_distance_pct=distance('long_target'), |
| 180 | lifecycle_reasons=reasons) |