Spaces:
Running
Running
| import os, re, json | |
| from datetime import datetime | |
| def safe_float_conversion(value, default=0.0): | |
| try: | |
| if value is None: return default | |
| if isinstance(value, (int, float)): return float(value) | |
| if isinstance(value, str): | |
| cleaned = ''.join(c for c in value if c.isdigit() or c in '.-') | |
| return float(cleaned) if cleaned else default | |
| return default | |
| except (ValueError, TypeError): return default | |
| def _apply_patience_logic(decision, hold_minutes, trade_data, processed_data): | |
| action = decision.get('action') | |
| if action == "CLOSE_TRADE" and hold_minutes < 20: | |
| current_price = processed_data.get('current_price', 0) | |
| entry_price = trade_data.get('entry_price', 0) | |
| try: profit_loss_percent = ((current_price - entry_price) / entry_price) * 100 | |
| except (TypeError, ZeroDivisionError): profit_loss_percent = 0 | |
| if profit_loss_percent < 2: | |
| decision.update({ | |
| 'action': "HOLD", | |
| 'reasoning': f"Patience Filter: Blocked premature sell. Held for {hold_minutes:.1f}m. Giving trade more time." | |
| }) | |
| return decision | |
| def parse_json_from_response(response_text: str): | |
| try: | |
| json_match = re.search(r'```json\n(.*?)\n```', response_text, re.DOTALL) | |
| if json_match: return json_match.group(1).strip() | |
| json_match = re.search(r'\{.*\}', response_text, re.DOTALL) | |
| if json_match: return json_match.group() | |
| return None | |
| except Exception: return None | |
| def validate_required_fields(data_dict: dict, required_fields: list) -> bool: | |
| return all(field in data_dict for field in required_fields) | |
| def format_technical_indicators(advanced_indicators): | |
| if not advanced_indicators: return "No data for advanced indicators." | |
| summary = [] | |
| for timeframe, indicators in advanced_indicators.items(): | |
| if indicators: | |
| parts = [] | |
| if 'rsi' in indicators: parts.append(f"RSI: {indicators['rsi']:.2f}") | |
| if 'macd_hist' in indicators: parts.append(f"MACD Hist: {indicators['macd_hist']:.4f}") | |
| if 'volume_ratio' in indicators: parts.append(f"Volume: {indicators['volume_ratio']:.2f}x") | |
| if parts: summary.append(f"{timeframe}: {', '.join(parts)}") | |
| return "\n".join(summary) if summary else "Insufficient indicator data." | |
| def format_strategy_scores(strategy_scores, recommended_strategy): | |
| if not strategy_scores: return "No strategy data available." | |
| summary = [f"Recommended Strategy: {recommended_strategy}"] | |
| sorted_scores = sorted(strategy_scores.items(), key=lambda item: item[1], reverse=True) | |
| for strategy, score in sorted_scores: | |
| score_display = f"{score:.3f}" if isinstance(score, (int, float)) else str(score) | |
| summary.append(f" • {strategy}: {score_display}") | |
| return "\n".join(summary) | |
| def local_analyze_opportunity(candidate_data): | |
| score = candidate_data.get('enhanced_final_score', candidate_data.get('final_score', 0)) | |
| quality_warnings = candidate_data.get('quality_warnings', []) | |
| rsi_critical = any('🚨 RSI CRITICAL' in warning for warning in quality_warnings) | |
| rsi_warning = any('⚠️ RSI WARNING' in warning for warning in quality_warnings) | |
| if rsi_critical: | |
| return { | |
| "action": "HOLD", "reasoning": "Local analysis: CRITICAL RSI levels - extreme overbought condition.", | |
| "trade_type": "NONE", "stop_loss": None, "take_profit": None, "expected_target_minutes": 15, | |
| "confidence_level": 0.1, "model_source": "local_safety_filter", "strategy": "GENERIC" | |
| } | |
| advanced_indicators = candidate_data.get('advanced_indicators', {}) | |
| if not advanced_indicators: | |
| return { | |
| "action": "HOLD", "reasoning": "Local analysis: Insufficient advanced indicator data.", | |
| "trade_type": "NONE", "stop_loss": None, "take_profit": None, "expected_target_minutes": 15, | |
| "confidence_level": 0.3, "model_source": "local", "strategy": "GENERIC" | |
| } | |
| action, reasoning, trade_type = "HOLD", "Local analysis: No strong buy signal based on enhanced rules.", "NONE" | |
| stop_loss, take_profit, expected_minutes, confidence = None, None, 15, 0.3 | |
| five_minute_indicators = advanced_indicators.get('5m', {}) | |
| one_hour_indicators = advanced_indicators.get('1h', {}) | |
| buy_conditions = total_conditions = 0 | |
| if isinstance(score, (int, float)) and score > 0.70: buy_conditions += 1 | |
| total_conditions += 1 | |
| rsi_five_minute = five_minute_indicators.get('rsi', 50) | |
| if 30 <= rsi_five_minute <= 65: buy_conditions += 1 | |
| total_conditions += 1 | |
| if five_minute_indicators.get('macd_hist', 0) > 0: buy_conditions += 1 | |
| total_conditions += 1 | |
| if (five_minute_indicators.get('ema_9', 0) > five_minute_indicators.get('ema_21', 0) and | |
| one_hour_indicators.get('ema_9', 0) > one_hour_indicators.get('ema_21', 0)): buy_conditions += 1 | |
| total_conditions += 1 | |
| if five_minute_indicators.get('volume_ratio', 0) > 1.5: buy_conditions += 1 | |
| total_conditions += 1 | |
| confidence = buy_conditions / total_conditions if total_conditions > 0 else 0.3 | |
| if rsi_warning: | |
| confidence *= 0.7 | |
| reasoning += " RSI warning applied." | |
| if confidence >= 0.6: | |
| action = "BUY" | |
| current_price = candidate_data['current_price'] | |
| trade_type = "LONG" | |
| stop_loss = current_price * 0.93 if rsi_warning else current_price * 0.95 | |
| take_profit = five_minute_indicators.get('bb_upper', current_price * 1.05) * 1.02 | |
| expected_minutes = 10 if confidence >= 0.8 else 18 if confidence >= 0.6 else 25 | |
| reasoning = f"Local enhanced analysis: Strong buy signal with {buy_conditions}/{total_conditions} conditions met. Confidence: {confidence:.2f}" | |
| if rsi_warning: reasoning += " (RSI warning - trading with caution)" | |
| return { | |
| "action": action, "reasoning": reasoning, "trade_type": trade_type, "stop_loss": stop_loss, | |
| "take_profit": take_profit, "expected_target_minutes": expected_minutes, "confidence_level": confidence, | |
| "model_source": "local", "strategy": "GENERIC" | |
| } | |
| def local_re_analyze_trade(trade_data, processed_data): | |
| current_price = processed_data['current_price'] | |
| stop_loss = trade_data['stop_loss'] | |
| take_profit = trade_data['take_profit'] | |
| action = "HOLD" | |
| reasoning = "Local re-analysis: No significant change to trigger an update or close." | |
| if stop_loss and current_price <= stop_loss: | |
| action, reasoning = "CLOSE_TRADE", "Local re-analysis: Stop loss has been hit." | |
| elif take_profit and current_price >= take_profit: | |
| action, reasoning = "CLOSE_TRADE", "Local re-analysis: Take profit has been hit." | |
| strategy = trade_data.get('strategy', 'GENERIC') | |
| if strategy == 'unknown': strategy = trade_data.get('decision_data', {}).get('strategy', 'GENERIC') | |
| return { | |
| "action": action, "reasoning": reasoning, "new_stop_loss": None, "new_take_profit": None, | |
| "new_expected_minutes": None, "model_source": "local", "strategy": strategy | |
| } |