301 lines
11 KiB
Python
Executable File
301 lines
11 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Generate Trading Signal - Combine quantitative analysis and LLM decision making
|
|
"""
|
|
import sys
|
|
import json
|
|
import logging
|
|
import os
|
|
import argparse
|
|
from pathlib import Path
|
|
|
|
# Add parent directory to path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent))
|
|
|
|
# Load .env file
|
|
from dotenv import load_dotenv
|
|
load_dotenv(Path(__file__).parent.parent / '.env')
|
|
|
|
from analysis.engine import MarketAnalysisEngine
|
|
from signals.quantitative import QuantitativeSignalGenerator
|
|
from signals.llm_decision import LLMDecisionMaker
|
|
from signals.aggregator import SignalAggregator
|
|
from notifiers.dingtalk import DingTalkNotifier
|
|
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def print_section(title: str, width: int = 80):
|
|
"""Print section header"""
|
|
print(f"\n{'=' * width}")
|
|
print(f"{title:^{width}}")
|
|
print(f"{'=' * width}")
|
|
|
|
|
|
def print_signal(signal: dict, title: str):
|
|
"""Pretty print a signal"""
|
|
print(f"\n{title}")
|
|
print("-" * 60)
|
|
print(f"Signal: {signal['signal_type']}")
|
|
print(f"Confidence: {signal.get('confidence', 0):.2%}")
|
|
|
|
# Display trade type if available (from LLM)
|
|
if 'trade_type' in signal:
|
|
trade_type = signal['trade_type']
|
|
trade_type_display = {
|
|
'INTRADAY': 'Intraday',
|
|
'SWING': 'Swing',
|
|
'NONE': 'None'
|
|
}.get(trade_type, trade_type)
|
|
print(f"Trade Type: {trade_type_display}")
|
|
|
|
if 'composite_score' in signal:
|
|
print(f"Composite Score: {signal['composite_score']:.1f}")
|
|
|
|
if 'scores' in signal:
|
|
print("\nComponent Scores:")
|
|
for component, score in signal['scores'].items():
|
|
print(f" {component:12}: {score:>6.1f}")
|
|
|
|
if 'levels' in signal:
|
|
levels = signal['levels']
|
|
print(f"\nPrice Levels:")
|
|
print(f" Current: ${levels.get('current_price', 0):>10,.2f}")
|
|
print(f" Entry: ${levels.get('entry', 0):>10,.2f}")
|
|
print(f" Stop: ${levels.get('stop_loss', 0):>10,.2f}")
|
|
print(f" Target 1: ${levels.get('take_profit_1', 0):>10,.2f}")
|
|
print(f" Target 2: ${levels.get('take_profit_2', 0):>10,.2f}")
|
|
print(f" Target 3: ${levels.get('take_profit_3', 0):>10,.2f}")
|
|
|
|
if 'risk_reward_ratio' in signal:
|
|
rr = signal['risk_reward_ratio']
|
|
if rr > 0:
|
|
print(f"\nRisk/Reward: 1:{rr:.2f}")
|
|
|
|
# Display opportunities breakdown (from LLM)
|
|
if 'opportunities' in signal:
|
|
opps = signal['opportunities']
|
|
|
|
# Intraday opportunity
|
|
if opps.get('intraday', {}).get('exists'):
|
|
intra = opps['intraday']
|
|
print(f"\nIntraday Opportunity:")
|
|
print(f" Direction: {intra.get('direction', 'N/A')}")
|
|
if intra.get('entry_price'):
|
|
print(f" Entry: ${intra['entry_price']:,.2f}")
|
|
if intra.get('stop_loss'):
|
|
print(f" Stop: ${intra['stop_loss']:,.2f}")
|
|
if intra.get('take_profit'):
|
|
print(f" Target: ${intra['take_profit']:,.2f}")
|
|
if intra.get('reasoning'):
|
|
print(f" Reasoning: {intra['reasoning']}")
|
|
|
|
# Swing opportunity
|
|
if opps.get('swing', {}).get('exists'):
|
|
swing = opps['swing']
|
|
print(f"\nSwing Opportunity:")
|
|
print(f" Direction: {swing.get('direction', 'N/A')}")
|
|
if swing.get('entry_price'):
|
|
print(f" Entry: ${swing['entry_price']:,.2f}")
|
|
if swing.get('stop_loss'):
|
|
print(f" Stop: ${swing['stop_loss']:,.2f}")
|
|
if swing.get('take_profit'):
|
|
print(f" Target: ${swing['take_profit']:,.2f}")
|
|
if swing.get('reasoning'):
|
|
print(f" Reasoning: {swing['reasoning']}")
|
|
|
|
if 'reasoning' in signal:
|
|
print(f"\nReasoning: {signal['reasoning']}")
|
|
|
|
|
|
def print_aggregated_signal(aggregated: dict):
|
|
"""Print aggregated signal"""
|
|
print_section("AGGREGATED TRADING SIGNAL")
|
|
|
|
print(f"\nFinal Signal: {aggregated['final_signal']}")
|
|
print(f"Confidence: {aggregated['final_confidence']:.2%}")
|
|
print(f"Consensus: {aggregated['consensus']}")
|
|
print(f"Agreement Score: {aggregated['agreement_score']:.2%}")
|
|
|
|
# Quantitative signal
|
|
print("\n" + "-" * 80)
|
|
quant = aggregated['quantitative_signal']
|
|
print(f"QUANTITATIVE SIGNAL: {quant.get('signal_type', quant.get('signal', 'HOLD'))} (confidence: {quant.get('confidence', 0):.2%})")
|
|
print(f" Composite Score: {quant.get('composite_score', 0):.1f}")
|
|
if 'scores' in quant:
|
|
scores = quant['scores']
|
|
print(f" Trend: {scores.get('trend', 0):>6.1f} | "
|
|
f"Momentum: {scores.get('momentum', 0):>6.1f} | "
|
|
f"OrderFlow: {scores.get('orderflow', 0):>6.1f} | "
|
|
f"Breakout: {scores.get('breakout', 0):>6.1f}")
|
|
|
|
# LLM signal
|
|
print("\n" + "-" * 80)
|
|
llm = aggregated.get('llm_signal')
|
|
if llm and isinstance(llm, dict):
|
|
trade_type_text = {
|
|
'INTRADAY': 'Intraday',
|
|
'SWING': 'Swing',
|
|
'AMBUSH': 'Ambush',
|
|
'NONE': 'None'
|
|
}.get(llm.get('trade_type', 'NONE'), llm.get('trade_type', 'N/A'))
|
|
|
|
print(f"LLM SIGNAL: {llm.get('signal_type', llm.get('signal', 'HOLD'))} (confidence: {llm.get('confidence', 0):.2%})")
|
|
print(f" Trade Type: {trade_type_text}")
|
|
|
|
# Display opportunities if available
|
|
if 'opportunities' in llm:
|
|
opps = llm['opportunities']
|
|
if opps.get('intraday', {}).get('exists'):
|
|
intra = opps['intraday']
|
|
print(f" Intraday: {intra.get('direction')} @ ${intra.get('entry_price', 0):,.0f}")
|
|
if opps.get('swing', {}).get('exists'):
|
|
swing = opps['swing']
|
|
print(f" Swing: {swing.get('direction')} @ ${swing.get('entry_price', 0):,.0f}")
|
|
|
|
print(f" Reasoning: {llm.get('reasoning', 'N/A')[:200]}")
|
|
if llm.get('key_factors'):
|
|
print(f" Key Factors: {', '.join(llm['key_factors'][:3])}")
|
|
else:
|
|
print("LLM SIGNAL: Not available (no API key configured)")
|
|
|
|
# Final levels
|
|
print("\n" + "-" * 80)
|
|
levels = aggregated['levels']
|
|
print("RECOMMENDED LEVELS:")
|
|
print(f" Current Price: ${levels['current_price']:>10,.2f}")
|
|
print(f" Entry: ${levels['entry']:>10,.2f}")
|
|
print(f" Stop Loss: ${levels['stop_loss']:>10,.2f}")
|
|
print(f" Take Profit 1: ${levels['take_profit_1']:>10,.2f}")
|
|
print(f" Take Profit 2: ${levels['take_profit_2']:>10,.2f}")
|
|
print(f" Take Profit 3: ${levels['take_profit_3']:>10,.2f}")
|
|
|
|
rr = aggregated.get('risk_reward_ratio', 0)
|
|
if rr > 0:
|
|
print(f"\n Risk/Reward Ratio: 1:{rr:.2f}")
|
|
|
|
# Recommendation
|
|
print("\n" + "-" * 80)
|
|
print(f"RECOMMENDATION:")
|
|
print(f" {aggregated['recommendation']}")
|
|
|
|
# Warnings
|
|
if aggregated.get('warnings'):
|
|
print("\n" + "-" * 80)
|
|
print("WARNINGS:")
|
|
for warning in aggregated['warnings']:
|
|
print(f" {warning}")
|
|
|
|
print("\n" + "=" * 80)
|
|
|
|
|
|
def main():
|
|
# Parse command line arguments
|
|
parser = argparse.ArgumentParser(description='Generate trading signals')
|
|
parser.add_argument('--send-dingtalk', action='store_true',
|
|
help='Send notification to DingTalk')
|
|
args = parser.parse_args()
|
|
|
|
print_section("TRADING SIGNAL GENERATOR", 80)
|
|
|
|
# Initialize components
|
|
logger.info("Initializing analysis engine...")
|
|
engine = MarketAnalysisEngine()
|
|
|
|
logger.info("Initializing signal generators...")
|
|
quant_generator = QuantitativeSignalGenerator()
|
|
|
|
# Initialize DingTalk notifier if requested
|
|
dingtalk = None
|
|
if args.send_dingtalk:
|
|
dingtalk_webhook = os.getenv('DINGTALK_WEBHOOK')
|
|
dingtalk_secret = os.getenv('DINGTALK_SECRET')
|
|
dingtalk = DingTalkNotifier(
|
|
webhook_url=dingtalk_webhook,
|
|
secret=dingtalk_secret,
|
|
enabled=bool(dingtalk_webhook)
|
|
)
|
|
|
|
# Initialize LLM decision maker
|
|
# Use 'openai' provider - supports OpenAI, Deepseek, and other OpenAI-compatible APIs
|
|
llm_maker = LLMDecisionMaker(provider='openai')
|
|
|
|
# Step 1: Perform market analysis
|
|
print_section("1. MARKET ANALYSIS")
|
|
analysis = engine.analyze_current_market(timeframe='5m')
|
|
|
|
if 'error' in analysis:
|
|
print(f"Error: {analysis['error']}")
|
|
print("\nTip: Check your network connection to Binance API")
|
|
return
|
|
|
|
print(f"Analysis complete")
|
|
print(f" Price: ${analysis['current_price']:,.2f}")
|
|
print(f" Trend: {analysis['trend_analysis'].get('direction', 'unknown')}")
|
|
print(f" RSI: {analysis['momentum'].get('rsi', 0):.1f}")
|
|
print(f" MACD: {analysis['momentum'].get('macd_signal', 'unknown')}")
|
|
|
|
# Step 2: Generate quantitative signal
|
|
print_section("2. QUANTITATIVE SIGNAL")
|
|
quant_signal = quant_generator.generate_signal(analysis)
|
|
print_signal(quant_signal, "Quantitative Analysis")
|
|
|
|
# Step 3: Generate LLM decision
|
|
print_section("3. LLM DECISION")
|
|
|
|
llm_context = engine.get_llm_context(format='full')
|
|
llm_signal = llm_maker.generate_decision(llm_context, analysis)
|
|
|
|
if llm_signal.get('enabled', True):
|
|
print_signal(llm_signal, "LLM Analysis")
|
|
else:
|
|
print("\nLLM Analysis: Disabled (no API key)")
|
|
print(" Set ANTHROPIC_API_KEY or OPENAI_API_KEY to enable")
|
|
|
|
# Step 4: Aggregate signals
|
|
print_section("4. SIGNAL AGGREGATION")
|
|
aggregated = SignalAggregator.aggregate_signals(quant_signal, llm_signal)
|
|
|
|
print_aggregated_signal(aggregated)
|
|
|
|
# Step 5: Export to JSON
|
|
output_file = Path(__file__).parent.parent / 'output' / 'latest_signal.json'
|
|
output_file.parent.mkdir(exist_ok=True)
|
|
|
|
output_data = {
|
|
'aggregated_signal': aggregated,
|
|
'market_analysis': {
|
|
'price': analysis['current_price'],
|
|
'trend': analysis['trend_analysis'],
|
|
'momentum': analysis['momentum'],
|
|
},
|
|
'quantitative_signal': quant_signal,
|
|
'llm_signal': llm_signal if llm_signal and llm_signal.get('enabled', True) else None,
|
|
}
|
|
|
|
with open(output_file, 'w') as f:
|
|
json.dump(output_data, f, indent=2, ensure_ascii=False)
|
|
|
|
print(f"\nSignal saved to: {output_file}")
|
|
|
|
# Send DingTalk notification if enabled
|
|
if dingtalk:
|
|
print(f"\nSending DingTalk notification...")
|
|
success = dingtalk.send_signal(aggregated)
|
|
if success:
|
|
print(f"DingTalk notification sent successfully")
|
|
else:
|
|
print(f"Failed to send DingTalk notification")
|
|
|
|
print_section("SIGNAL GENERATION COMPLETE", 80)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|