edgeyopt/solver/solver.py

'''
'''
from data import series
from execution.executor import Executor

import random
from math import inf
import numpy
from PyQt5.Qt import pyqtSignal, QObject
from PyQt5 import QtCore
import talib
from data.signal import PriceComparisonSignal, DayOfMonthSignal, DayOfWeekSignal,\
    StochasticSignal, IntradayBarNumberSignal, BbandsSignal,\
    GainPercentrankSignal

class Solver(QObject):
    '''
    
    '''
    
    progress = pyqtSignal(int, int, name='progress')
    done = pyqtSignal(list, name='done')

    def __init__(self, series):
        '''
        Constructor
        '''
        super().__init__(None)
        
        self.series = series
        self.generators = []
        self.counter = 0
        self.total_counter = 0
        
    def add_generator(self, generator):
        self.generators.append(generator)
    
    @QtCore.pyqtSlot(dict)
    def solve(self, params):
              
        max_signals = 5
        max_hold_bars = params.get('max_hold_bars', 1)
        self.executor = Executor(self.series, max_hold_bars)
        max_strategies = params.get('num_strategies', 1000)
        results = []
        
        min_trades = params.get('min_trades', 0)
        min_win_rate = params.get('min_win_rate', 0)
        min_sharpe = params.get('min_sharpe', 0)
        
        stop = params.get('stop_loss', None)
        tp = params.get('take_profit', None)
        
        is_long = params.get('direction', 'long') == 'long'
        
        strategy = [GainPercentrankSignal(107, 1, GainPercentrankSignal.LT)]
        signals = self.make_signals(strategy)
                
        trades = self.executor.execute(signals, is_long, stop, tp)
        result = self.evaluate_trades(trades)
        result['strategy'] = strategy
        result['display_name'] = ' && '.join([signal.get_text() for signal in strategy])
        result['trades'] = trades
        results.append(result)
        self.progress.emit(len(results), max_strategies)
        
        self.done.emit([result])
        
        while len(results) < max_strategies:
            sig_num = random.randint(1, max_signals)
            strategy = []
            for i in range(0, sig_num):
                strategy.append(random.choice(self.generators).generate())
                
            signals = self.make_signals(strategy)
                
            trades = self.executor.execute(signals, is_long, stop, tp)
            if len(trades) >= min_trades:
                result = self.evaluate_trades(trades)
                if result['win_percentage'] > min_win_rate and result['sharpe'] > min_sharpe:
                    result['strategy'] = strategy
                    result['display_name'] = ' && '.join([signal.get_text() for signal in strategy])
                    result['trades'] = trades
                    results.append(result)
                    self.progress.emit(len(results), max_strategies)
                    
                    self.done.emit([result])
                    
            self.counter += 1
            self.total_counter += 1
            
    def make_signals(self, strategy):
        sig_vectors = numpy.ones(self.series.length(), dtype=int)
        for signal in strategy:
            vec = signal.calculate(self.series)
            assert(len(vec) == self.series.length())
            sig_vectors = numpy.logical_and(sig_vectors, vec)
        
        return sig_vectors
            
    def evaluate_trades(self, trades):
        result = {}
        
        profits = [x.pnl() for x in trades]
        
        total_won = len(list(filter(lambda x: x.pnl() > 0, trades)))

        if len(trades) > 0:
            result['win_percentage'] = total_won / len(trades) * 100
        else:
            result['win_percentage'] = 0
        
        result['trades_number'] = len(trades)
        result['total_pnl'] = sum(profits)
        
        if len(trades) > 0:
            result['avg_percentage'] = sum([trade.pnl_percentage() for trade in trades]) / len(trades)
        else:
            result['avg_percentage'] = 0
        
        gross_profit = sum([max(0, x.pnl()) for x in trades])
        gross_loss = sum([min(0, x.pnl()) for x in trades])
        
        if gross_loss != 0:
            result['profit_factor'] = gross_profit / (-gross_loss)
        else:
            result['profit_factor'] = inf
            
        if len(profits) > 0:
            mean = numpy.mean(profits)
            stddev = numpy.std(profits)
            if stddev != 0:
                result['sharpe'] = mean / stddev
            else:
                result['sharpe'] = 0
        else:
            result['sharpe'] = 0
            
        return result