text:
#!/usr/bin/env python3
# python libraries
import os, sys, shutil, copy
import matplotlib.pyplot as plt
import numpy as np
# Peter Lepage's libraries
import gvar as gv
import lsqfit

# FK / Fpi libraries
sys.path.append('utils')
import io_utils
import chipt
import analysis
import plotting

''' Write description
'''

''' useful for changing matplotlib.plt.show behavior and accessing results
    from main() if run interactively in iPython
'''
def run_from_ipython():
    try:
        __IPYTHON__
        return True
    except NameError:
        return False

def main():
    print("python     version:", sys.version)
    import input_params as ip
    # Load input params
    switches   = ip.switches
    priors     = ip.priors
    phys_point = ip.phys_point
    check_fit  = ip.check_fit

    # if check_fit: - add support
    if switches['check_fit']:
        models = analysis.sys_models(switches)
        print('DEBUGGING FIT FUNCTION')
        print('p')
        for k in check_fit['p']:
            print("%7s" %k,check_fit['p'][k])
        print('x')
        for k in check_fit['x']:
            print("%7s" %k,check_fit['x'][k])
        for model in models:
            print('===============================================================')
            print('DEBUGGING Terms in ',model)
            print('---------------------------------------------------------------')
            model_list, FF, fv = analysis.gather_model_elements(model)
            debug_fit_function(check_fit, model_list, FF, fv)
        sys.exit()
    if switches['print_Li']:
        li_rho = r'$10^3L_i(m_\rho)$& '
        li_mu0 = r'$10^3L_i(\mu_0)$& '
        for Li in ['L1','L2','L3','L4','L5','L6','L7','L8']:
            li_rho += str(ip.Li_mrho[Li])+'& '
            li_mu0 += str(1e3*ip.priors[Li])+'& '
        print(li_rho)
        print(li_mu0)
        sys.exit('Exiting after printing L_i values')

    if switches['save_fits']:
        if not os.path.exists('pickled_fits'):
            os.makedirs('pickled_fits')

    # load data
    gv_data = io_utils.format_h5_data('data/FK_Fpi_data.h5',switches)

    models = analysis.sys_models(switches)
    if switches['prior_search']:
        print('Performing Prior Width Scan')
        print(r'%33s &  nnlo_x &  nnlo_a &  n3lo_x &  n3lo_a & logGBF_max' %'model')
        print('======================================================================================')
        for model in models:
            model_list, FF, fv = analysis.gather_model_elements(model)
            fit_model  = chipt.FitModel(model_list, _fv=fv, _FF=FF)
            fitEnv     = FitEnv(gv_data, fit_model, switches)
            analysis.prior_width_scan(model, fitEnv, fit_model, priors, switches)
    else: # do analysis
        fit_results = dict()
        plt.ion()
        for model in models:
            print('===============================================================')
            print(model)
            model_list, FF, fv = analysis.gather_model_elements(model)
            fit_model  = chipt.FitModel(model_list, _fv=fv, _FF=FF)
            fitEnv     = FitEnv(gv_data, fit_model, switches)

            do_fit = False
            pickled_fit = 'pickled_fits/'+model+'_n2lo'+str(ip.nnlo_x)+'_n3lo'+str(ip.n3lo_x)+'.p'
            if switches['save_fits'] or switches['debug_save_fit']:
                if os.path.exists(pickled_fit):
                    print('reading %s' %pickled_fit)
                    fit_result = gv.load(pickled_fit)
                    check_pickled_fit(fit_result,switches,priors)
                else:
                    do_fit = True
            else:
                do_fit = True
            if do_fit or switches['debug_save_fit']:
                tmp_result = fitEnv.fit_data(priors)
                tmp_result.phys_point = dict()
                tmp_result.phys_point.update({k:v for k,v in phys_point['p'].items() if ('Lchi' in k) or k in ['mpi','mk','mkp']})
                if switches['debug_save_fit']:
                    print('live fit')
                    report_phys_point(tmp_result, phys_point, model_list, FF, report=True)
                    analysis.uncertainty_breakdown(tmp_result,'FKFpi',print_error=True)
                    print('pickled fit')
                    if not os.path.exists(pickled_fit):
                        gv.dump(tmp_result, pickled_fit, add_dependencies=True)
                        fit_result = gv.load(pickled_fit)
                    report_phys_point(fit_result, phys_point, model_list, FF, report=True)
                    analysis.uncertainty_breakdown(fit_result,'FKFpi',print_error=True)
                if do_fit:
                    fit_result = tmp_result

            if switches['print_fit']:
                print(fit_result.format(maxline=True))
            fit_result.phys_point.update({k:v for k,v in phys_point['p'].items() if ('Lchi' in k) or k in ['mpi','mk','mkp']})
            fit_result.ensembles_fit = switches['ensembles_fit']
            report_phys_point(fit_result, phys_point, model_list, FF, report=switches['report_phys'])
            if switches['report_Li']:
                for Li in ['L1','L2','L3','L4','L5','L6','L7','L8']:
                    if Li in fit_result.p:
                        Li_mu0 = fit_result.p[Li]
                        Li_rho = Li_mu0 - ip.gamma_i[Li]/(4*np.pi)**2 * np.log(770 / (4*np.pi*80))
                        print("%s   %10s   %10s" %(Li, 1e3*Li_mu0, 1e3*Li_rho))
                for k in ['k_4', 'p_4', 'kp_6', 'k_6', 'p_6']:
                    if k in fit_result.p:
                        print("%4s   %s" %(k,fit_result.p[k]))
            fit_results[model] = fit_result
            if switches['save_fits']:
                gv.dump(fit_result, pickled_fit, add_dependencies=True)

            if switches['debug_phys_point'] and not do_fit:
                report_phys_point(fit_result, phys_point, model_list, FF)
            if switches['make_extrap'] or switches['make_fv']:
                plots = plotting.ExtrapolationPlots(model, model_list, fitEnv, fit_result, switches)
            if switches['make_extrap']:
                if 'alphaS' not in model and 'ma' not in model:
                    plots.plot_vs_eps_asq(phys_point)
                if 'ma' not in model:
                    plots.plot_vs_eps_pi(phys_point)
            if switches['make_fv']:
                if 'xpt' in model and 'FV' in model and 'PP' in model:
                    plots.plot_vs_ml()

        if switches['model_avg']:
            model_avg = analysis.BayesModelAvg(fit_results)
            model_avg.print_weighted_models()
            model_avg.bayes_model_avg()
            if switches['make_hist']:
                model_avg.plot_bma_hist('FF',save_fig=switches['save_figs'])
                model_avg.plot_bma_hist('FF_xpt',save_fig=switches['save_figs'])
                model_avg.plot_bma_hist('ratio',save_fig=switches['save_figs'])
                model_avg.plot_bma_hist('ct',save_fig=switches['save_figs'])

        plt.ioff()
        if run_from_ipython():
            plt.show(block=False)
            return fit_results
        else:
            plt.show()

        return fit_results

'''
    This is the main class that runs a given fit specified by a model
'''
class FitEnv:
    # xyp_dict is a dictionary with keys 'x', 'y', 'p'
    # 'y' value is a dict{ensemble : yval},
    # 'x' value is a dict{ensemble : {<all the x variables which are not priors>}
    # 'p' value is a dict{(ensemble, pKey) : aPal} for all the
    #       ensemble-specific priors like meson masses as well as teh LECs
    def __init__(self, xyp_dict, model, switches):
        self.switches   = switches
        self.ensembles  = switches['ensembles_fit']
        self.x          = xyp_dict['x']
        self.pruned_x   = {ens : { k : v for k, v in xyp_dict['x'][ens].items()}
                                for ens in self.ensembles}
        self.y          = xyp_dict['y']
        self.pruned_y   = {ens : xyp_dict['y'][ens] for ens in self.ensembles}
        required_params = model.get_required_parameters()
        self.p          = xyp_dict['p']
        self.pruned_p   = {(ens, k) : v for (ens, k), v in xyp_dict['p'].items()
                                if k in required_params and ens in self.ensembles}
        self.model      = model

    # create a callable function that acts on a single x and p (not all ensembles)
    @classmethod
    def _fit_function(cls, a_model, x, p):
        return a_model(x,p)

    def fit_function(self, x, p):
        a_result = dict()
        for ens in x.keys():
            p_ens = dict()
            for k, v in p.items():
                if type(k) == tuple and k[0] == ens:
                    p_ens[k[1]] = v # the x-params which are priors
                else:
                    p_ens[k] = v    # the LECs of the fit
            model = self.model
            a_result[ens] = FitEnv._fit_function(model, x[ens], p_ens)
        return a_result

    def fit_data(self, lec_priors):
        required_params = self.model.get_required_parameters()
        # add the LEC priors to our list of priors for the fit
        self.pruned_p.update({ k:v for k,v in lec_priors.items() if k in required_params})
        x = self.pruned_x
        y = self.pruned_y
        p = self.pruned_p
        if self.switches['scipy']:
            fitter='scipy_least_squares'
        else:
            fitter='gsl_multifit'
        return lsqfit.nonlinear_fit(data=(x,y), prior=p, fcn=self.fit_function, fitter=fitter, debug=True)

def report_phys_point(fit_result, phys_point_params, model_list, FF, report=False):
    phys_data = copy.deepcopy(phys_point_params)
    if 'ma' in model_list[0]:
        model_list[0] = model_list[0].replace('ma','xpt')
    fit_model = chipt.FitModel(model_list, _fv=False, _FF=FF)
    for k in fit_result.p:
        if isinstance(k,str):
            phys_data['p'][k] = fit_result.p[k]
    fit_result.phys                = dict()
    fit_result.phys['FKFpi']       = FitEnv._fit_function(fit_model, phys_data['x'], phys_data['p'])
    fit_result.phys['dF_iso_xpt']  = chipt.dFKFpi_iso(phys_data, FF)
    fit_result.phys['dF_iso_xpt2'] = chipt.dFKFpi_iso_2(phys_data, FF, fit_result.phys['FKFpi'])
    if report:
        print('  chi2/dof [dof] = %.2f [%d]   Q=%.3f   logGBF = %.3f' \
            %(fit_result.chi2/fit_result.dof, fit_result.dof, fit_result.Q, fit_result.logGBF))
        print('  FK/Fpi                = %s' %fit_result.phys['FKFpi'])
        print('  dF_iso_xpt            = %s' %chipt.dFKFpi_iso(phys_data, FF))
        print('  dF_iso_xpt2           = %s' %chipt.dFKFpi_iso_2(phys_data, FF, fit_result.phys['FKFpi']))
        print('  dF_iso_vincenzo       = %s' %chipt.dFKFpi_vincenzo(phys_data, FF))
        print('  dF_iso_vincenzo2      = %s' %chipt.dFKFpi_vincenzo_2(phys_data, FF, fit_result.phys['FKFpi']))

def debug_fit_function(check_fit, model_list, FF, fv):
    x = check_fit['x']
    p = check_fit['p']
    fit_model = chipt.FitModel(model_list, _fv=False, _FF=FF)
    cP        = chipt.ConvenienceDict(fit_model, x, p)
    result    = 0.
    result_FV = 0.
    if fv:
        fit_model_fv = chipt.FitModel(model_list, _fv=True, _FF=FF)
        cP_FV        = chipt.ConvenienceDict(fit_model_fv, x, p)
        for term in model_list:
            if '_nlo' in term:
                t_FV = getattr(chipt.FitModel, term)(fit_model_fv, x, p, cP_FV)
                t    = getattr(chipt.FitModel, term)(fit_model, x, p, cP)
                result    += t
                result_FV += t_FV
                print('%16s   ' %(term+'_FV'), t_FV)
                print('%16s   ' %(term), t)
            else:
                t = getattr(chipt.FitModel, term)(fit_model, x, p, cP)
                result_FV += t
                result    += t
                print('%16s   ' %(term), t)
    else:
        for term in model_list:
            t = getattr(chipt.FitModel, term)(fit_model, x, p, cP)
            result += t
            print('%16s   ' %(term), t)
    print('---------------------------------------')
    if fv:
        print('%16s   %f' %('total_FV', result_FV))
    print('%16s   %f' %('total', result))

def check_pickled_fit(fit,switches,priors):
    ''' make sure the pickled data is consistent with choices in switches '''
    if not set(fit.ensembles_fit) == set(switches['ensembles_fit']):
        sys.exit('ensembles_fit from the pickled fit does not match ensembles_fit from input_params')
    for p in priors:
        if p in fit.prior:
            if priors[p].mean != fit.prior[p].mean or priors[p].sdev != fit.prior[p].sdev:
                sys.exit('prior %s from fit, %s, does not match from input_params %s' \
                    %(p,fit.prior[p],priors[p]))

if __name__ == "__main__":
    main()


text:
import unittest
import numpy as np
import os
from dama.data.ds import Data
from dama.drivers.core import HDF5
from dama.utils.model_selection import CV
from dama.ae.extended.w_keras import PTsne
from dama.models import MetadataX
from dama.utils.tf_functions import TSNe
from dama.data.it import BatchIterator
from dama.utils.files import check_or_create_path_dir
from dama.utils.core import Chunks
from dama.connexions.core import ListConn

TMP_PATH = check_or_create_path_dir(os.path.dirname(os.path.abspath(__file__)), 'dama_data_test')

def list_conn(iterator, dtypes):
    for it in iterator:
        list_conn = ListConn([], dtypes)
        list_conn[0] = it[0]
        list_conn[1] = it[1]
        yield list_conn

class TestUnsupervicedModel(unittest.TestCase):
    def setUp(self):
        np.random.seed(0)
        x = np.random.rand(120)
        self.x = np.sin(6 * x).reshape(-1, 1)
        self.hash = None

    def tearDown(self):
        pass

    def train(self, ae, model_params=None):
        with Data(name="tsne", driver=HDF5(mode="w", path=TMP_PATH), metadata_path=TMP_PATH) as dataset, \
                Data(name="test_tsne", driver=HDF5(mode="w", path=TMP_PATH), metadata_path=TMP_PATH) as ds:
            batch_size = 12
            tsne = TSNe(batch_size=batch_size, perplexity=ae.perplexity, dim=2)
            dtypes = np.dtype([("x", np.dtype(float)), ("y", np.dtype(float))])
            x_p = BatchIterator.from_batchs(list_conn(tsne.calculate_P(self.x), dtypes), length=len(self.x),
                                            from_batch_size=tsne.batch_size, dtypes=dtypes, to_slice=True)
            dataset.from_data(x_p)
            cv = CV(group_data="x", group_target="y", train_size=.7, valid_size=.1)
            stc = cv.apply(dataset)
            ds.from_data(stc)
            ae.train(ds, num_steps=5, data_train_group="train_x", target_train_group='train_y', batch_size=tsne.batch_size,
                      data_test_group="test_x", target_test_group='test_y', model_params=model_params,
                      data_validation_group="validation_x", target_validation_group="validation_y")
            ae.save("tsne", path=TMP_PATH, model_version="1")
            dataset.destroy()
        return ae

    def test_parametric_tsne(self):
        ae = self.train(PTsne(metadata_path=TMP_PATH), model_params=None)
        metadata = MetadataX.get_metadata(ae.path_metadata_version)
        self.assertEqual(len(metadata["train"]["model_json"]) > 0, True)
        with Data(name="test0", driver=HDF5(mode="w", path=TMP_PATH), metadata_path=TMP_PATH) as dataset:
            dataset.from_data(self.x)

        with PTsne.load(model_version="1", model_name="tsne", path=TMP_PATH, metadata_path=TMP_PATH) as ae,\
                Data(name="test0", driver=HDF5(mode="r", path=TMP_PATH),
                     chunks=Chunks({"g0": (10, 1)}), metadata_path=TMP_PATH) as dataset:
            self.assertEqual(ae.predict(dataset).shape, (120, 2))
            ae.destroy()
            dataset.destroy()
            self.hash = ae.ds.hash

    def test_P(self):
        tsne = TSNe(batch_size=12, perplexity=30, dim=2)
        x_p = BatchIterator.from_batchs(tsne.calculate_P(self.x), length=len(self.x), from_batch_size=tsne.batch_size,
                                        dtypes=np.dtype([("x", np.dtype(float)), ("y", np.dtype(float))]))
        for i, e in enumerate(x_p):
            if i < 10:
                self.assertEqual(e[0].shape, (12, 1))
                self.assertEqual(e[1].shape, (12, 12))
            else:
                self.assertEqual(e[0].shape, (4, 1))
                self.assertEqual(e[1].shape, (4, 12))

if __name__ == '__main__':
    unittest.main()

ext: py

max_stars_repo_name: elaeon/ML

reason: import numpy


text:
import sys,csv,re,math
import numpy as np

B=10000

metadata={}
docIdsByYear={}
knownAuthorsByYear={}
docIdsByYearWithKnownAuthor={}
seenDocs={}

genderdict={"f": 0, "m": 1}
roledict={"agent": 0, "patient": 1, "mod": 2, "poss": 3, "speaking": 4, "characters": 5}
data={}

def readmeta(filename):
	seen={}
	with open(filename, 'rb') as f:
		reader = csv.reader(f)
		next(reader)
		for row in reader:
			docid=row[0]
			date=int(row[5])
			authgender=row[7]
			author=row[3]
			title=row[9]
			enumcron=row[8]

			if date not in knownAuthorsByYear:
				knownAuthorsByYear[date]=[]

			if authgender != "u" and docid not in seen:
				knownAuthorsByYear[date].append(docid)
				seen[docid]=1

			metadata[docid]=(date, authgender, title, author)

def read(filenames):
	for filename in filenames:
		with open(filename, 'rb') as f:
			reader = csv.reader(f)
			next(reader)
			for row in reader:
				docid=row[0]
				date=int(row[1])
				gender=row[2]
				role=row[3]
				count=int(row[4])

				authgender='u'
				if docid in metadata:
					(date, authgender, title, author)=metadata[docid]

				if date not in docIdsByYear:
					docIdsByYear[date]=[]
					docIdsByYearWithKnownAuthor[date]=[]

				# skip if character gender is unknown
				if gender == "u":
					continue

				if docid not in seenDocs:
					seenDocs[docid]=1
					docIdsByYear[date].append(docid)

					if authgender != 'u':
						docIdsByYearWithKnownAuthor[date].append(docid)

				if docid not in data:
					data[docid]=np.zeros((2,6))

				# multivolume works appear as multiple books with the same docid, so append counts rather than assign
				data[docid][genderdict[gender],roledict[role]]+=count

def resampleAuthorCharacterCounts(year):

	resample=np.random.choice(docIdsByYearWithKnownAuthor[year], len(docIdsByYearWithKnownAuthor[year]))
	counts=np.zeros((2,2))
	for docid in resample:
		(date, authorgender, title, author)=metadata[docid]

		datum=data[docid]

		counts[genderdict[authorgender],0]+=np.sum(datum[0,0:4])
		counts[genderdict[authorgender],1]+=np.sum(datum[1,0:4])

	w_author=counts[0,0]/np.sum(counts[0,])
	m_author=counts[1,0]/np.sum(counts[1,])

	return w_author, m_author

def resampleAuthor(year):

	resample=np.random.choice(knownAuthorsByYear[year], len(knownAuthorsByYear[year]))

	counts=np.zeros(2)
	for docid in resample:
		(date, authorgender, title, author)=metadata[docid]

		counts[genderdict[authorgender]]+=1

	return counts[0]/np.sum(counts)

def resampleNumCharacter(year):

	resample=np.random.choice(docIdsByYear[year], len(docIdsByYear[year]))
	counts=np.zeros(2)
	for docid in resample:
		datum=data[docid]

		counts[0]+=datum[0,5]
		counts[1]+=datum[1,5]

	return counts[0]/np.sum(counts)

def resampleSpeech(year):

	resample=np.random.choice(docIdsByYear[year], len(docIdsByYear[year]))
	counts=np.zeros(2)
	for docid in resample:
		datum=data[docid]

		# only use agent + patient + mod + poss counts
		counts[0]+=datum[0,4]
		counts[1]+=datum[1,4]

	return counts[0]/np.sum(counts)

def resampleTotalCounts(year):

	resample=np.random.choice(docIdsByYear[year], len(docIdsByYear[year]))
	counts=np.zeros(2)
	for docid in resample:
		datum=data[docid]

		# only use agent + patient + mod + poss counts
		counts[0]+=np.sum(datum[0,0:4])
		counts[1]+=np.sum(datum[1,0:4])

	return counts[0]/np.sum(counts)

def resampleYearDouble(year, samplingMethod):
	w_means=[]
	m_means=[]
	for b in range(B):
		(w_mean, m_mean)=samplingMethod(year)
		if not math.isnan(w_mean):
			w_means.append(w_mean)
		if not math.isnan(m_mean):
			m_means.append(m_mean)

	print "%s\t%s\t%s" % (year, '\t'.join("%.3f" % x for x in np.percentile(w_means, [2.5, 50, 97.5])), '\t'.join("%.3f" % x for x in np.percentile(m_means, [2.5, 50, 97.5])))

def resampleYear(year, samplingMethod):
	means=[]
	for b in range(B):
		mean=samplingMethod(year)
		means.append(mean)
	print "%s\t%s" % (year, '\t'.join("%.3f" % x for x in np.percentile(means, [2.5, 50, 97.5])))

# ex. Usage: python gender_plots.py fig1_ci filtered_fiction_metadata.csv post22_character_data.csv pre23_character_data.csv
chart=sys.argv[1]
readmeta(sys.argv[2])
read(sys.argv[3:])

if chart == "fig1_ci":
	for year in range(1800,2008):
		resampleYear(year, resampleTotalCounts)

elif chart == "author_ci":
	for year in range(1800,2008):
		resampleYear(year, resampleAuthor)

elif chart == "speech":
	for year in range(1800,2008):
		resampleYear(year, resampleSpeech)

elif chart == "numcharacter":
	for year in range(1800,2008):
		resampleYear(year, resampleNumCharacter)

elif chart == "authorcharacter":
	for year in range(1800,2008):
		resampleYearDouble(year, resampleAuthorCharacterCounts)

max_stars_repo_name: tedunderwood/character

max_issues_repo_name: tedunderwood/character

max_forks_repo_name: tedunderwood/character


text:
# coding: utf-8

"""
Automated Tool for Optimized Modelling (ATOM)
Author: Mavs
Description: Unit tests for atom.py

"""

import glob
from unittest.mock import patch

import numpy as np
import pandas as pd
import pytest
from sklearn.decomposition import PCA
from sklearn.ensemble import RandomForestClassifier
from sklearn.feature_selection import SelectFromModel
from sklearn.kernel_approximation import RBFSampler
from sklearn.linear_model import LassoLarsCV
from sklearn.metrics import get_scorer
from sklearn.neural_network import MLPClassifier
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import (
    LabelEncoder, OneHotEncoder, RobustScaler, StandardScaler,
)

from atom import ATOMClassifier, ATOMRegressor
from atom.data_cleaning import Pruner, Scaler
from atom.utils import check_scaling

from .utils import (
    FILE_DIR, X10, X10_dt, X10_nan, X10_str, X10_str2, X20_out, X_bin, X_class,
    X_reg, X_sparse, X_text, y10, y10_sn, y10_str, y_bin, y_class, y_reg,
)

# Test __init__ ==================================================== >>

def test_task_assignment():
    """Assert that the correct task is assigned."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    assert atom.task == "binary classification"

    atom = ATOMClassifier(X_class, y_class, random_state=1)
    assert atom.task == "multiclass classification"

    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    assert atom.task == "regression"

def test_raise_one_target_value():
    """Assert that error raises when there is only 1 target value."""
    y = [1 for _ in range(len(y_bin))]  # All targets are equal to 1
    pytest.raises(ValueError, ATOMClassifier, X_bin, y, random_state=1)

# Test magic methods =============================================== >>

def test_repr():
    """Assert that the __repr__ method visualizes the pipeline(s)."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.scale()
    assert "Branches: master" in str(atom)
    atom.branch = "b2"
    assert "Branches:\n   >>> master\n   >>> b2 !" in str(atom)

def test_iter():
    """Assert that we can iterate over atom's pipeline."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.clean()
    atom.impute()
    assert [item for item in atom][1] == atom.pipeline[1]

# Test utility properties =========================================== >>

def test_branch_same():
    """Assert that we can stay on the same branch."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    atom.branch = "master"
    assert atom.branch.name == "master"

def test_branch_change():
    """Assert that we can change to another branch."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    atom.branch = "b2"
    atom.clean()
    atom.branch = "master"
    assert atom.pipeline.empty  # Has no Cleaner

def test_branch_empty():
    """Assert that an error is raised when name is empty."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    with pytest.raises(ValueError, match=r".*have an empty name.*"):
        atom.branch = ""

def test_branch_existing_name():
    """Assert that an error is raised when the name already exists."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    atom.branch = "b2"
    with pytest.raises(ValueError, match=r".*already exists.*"):
        atom.branch = "b2_from_master"

def test_branch_model_acronym():
    """Assert that an error is raised when the name is a models' acronym."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    with pytest.raises(ValueError, match=r".*model's acronym.*"):
        atom.branch = "Lda"

def test_branch_unknown_parent():
    """Assert that an error is raised when the parent doesn't exist."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    with pytest.raises(ValueError, match=r".*does not exist.*"):
        atom.branch = "b2_from_invalid"

def test_branch_new():
    """Assert that we can create a new branch."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    atom.clean()
    atom.branch = "b2"
    assert list(atom._branches) == ["og", "master", "b2"]

def test_branch_from_valid():
    """Assert that we can create a new branch, not from the current one."""
    atom = ATOMClassifier(X10_nan, y10, random_state=1)
    atom.branch = "b2"
    atom.impute()
    atom.branch = "b3_from_master"
    assert atom.branch.name == "b3"
    assert atom.n_nans > 0

def test_scaled():
    """Assert that scaled returns if the dataset is scaled."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    assert not atom.scaled
    atom.scale()
    assert atom.scaled

def test_duplicates():
    """Assert that duplicates returns the number of duplicated samples."""
    atom = ATOMClassifier(X10, y10, random_state=1)
    assert atom.duplicates == 2

def test_nans():
    """Assert that nans returns a series of missing values."""
    atom = ATOMClassifier(X10_nan, y10, random_state=1)
    assert atom.nans.sum() == 2

def test_n_nans():
    """Assert that n_nans returns the number of rows with missing values."""
    atom = ATOMClassifier(X10_nan, y10, random_state=1)
    assert atom.n_nans == 2

def test_numerical():
    """Assert that numerical returns the names of the numerical columns."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    assert len(atom.numerical) == 3

def test_n_numerical():
    """Assert that n_categorical returns the number of numerical columns."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    assert atom.n_numerical == 3

def test_categorical():
    """Assert that categorical returns the names of categorical columns."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    assert len(atom.categorical) == 1

def test_n_categorical():
    """Assert that n_categorical returns the number of categorical columns."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    assert atom.n_categorical == 1

def test_outliers():
    """Assert that nans returns a series of outlier values."""
    atom = ATOMClassifier(X20_out, y10 * 2, random_state=1)
    assert atom.outliers.sum() == 2

def test_n_outliers():
    """Assert that n_outliers returns the number of rows with outliers."""
    atom = ATOMClassifier(X20_out, y10 * 2, random_state=1)
    assert atom.n_outliers == 2

def test_classes():
    """Assert that the classes property returns a df of the classes in y."""
    atom = ATOMClassifier(X_class, y_class, random_state=1)
    assert list(atom.classes.index) == [0, 1, 2]

def test_n_classes():
    """Assert that the n_classes property returns the number of classes."""
    atom = ATOMClassifier(X_class, y_class, random_state=1)
    assert atom.n_classes == 3

# Test utility methods ============================================= >>

@patch("tpot.TPOTClassifier")
def test_automl_classification(cls):
    """Assert that the automl method works for classification tasks."""
    pl = Pipeline(
        steps=[
            ("standardscaler", StandardScaler()),
            ("robustscaler", RobustScaler()),
            ("mlpclassifier", MLPClassifier(alpha=0.001, random_state=1)),
        ]
    )
    cls.return_value.fitted_pipeline_ = pl.fit(X_bin, y_bin)

    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("Tree", metric="accuracy")
    atom.branch = "automl"  # Change branch since new pipeline
    atom.automl()
    cls.assert_called_with(
        n_jobs=1,
        random_state=1,
        scoring=atom._metric[0],  # Called using atom's metric
        verbosity=0,
    )
    assert len(atom.pipeline) == 2
    assert atom.models == ["Tree", "MLP"]

@patch("tpot.TPOTRegressor")
def test_automl_regression(cls):
    """Assert that the automl method works for regression tasks."""
    pl = Pipeline(
        steps=[
            ("rbfsampler", RBFSampler(gamma=0.95, random_state=2)),
            ("lassolarscv", LassoLarsCV(normalize=False)),
        ]
    )
    cls.return_value.fitted_pipeline_ = pl.fit(X_reg, y_reg)

    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.automl(scoring="r2", random_state=2)
    cls.assert_called_with(n_jobs=1, scoring="r2", verbosity=0, random_state=2)
    assert atom.metric == "r2"

def test_automl_invalid_scoring():
    """Assert that an error is raised when the provided scoring is invalid."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.run("Tree", metric="mse")
    with pytest.raises(ValueError, match=r".*scoring parameter.*"):
        atom.automl(scoring="r2")

@pytest.mark.parametrize("distributions", [None, "norm", ["norm", "pearson3"]])
@pytest.mark.parametrize("columns", ["feature_1", 1, None])
def test_distribution(distributions, columns):
    """Assert that the distribution method and file are created."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    df = atom.distribution(distributions=distributions, columns=columns)
    assert isinstance(df, pd.DataFrame)

def test_export_pipeline_empty():
    """Assert that an error is raised when the pipeline is empty."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    pytest.raises(ValueError, atom.export_pipeline)

def test_export_pipeline_verbose():
    """Assert that the verbosity is passed to the transformers."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.clean()
    atom.run("LGB")
    assert atom.export_pipeline("LGB", verbose=2)[0].verbose == 2

def test_export_pipeline_same_transformer():
    """Assert that two same transformers get different names."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.clean()
    atom.clean()
    atom.clean()
    pl = atom.export_pipeline()
    assert list(pl.named_steps) == ["cleaner", "cleaner2", "cleaner3"]

def test_export_pipeline_invalid_model():
    """Assert that an error is raised when model is from other branch."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("GNB")
    atom.branch = "b2"
    pytest.raises(ValueError, atom.export_pipeline, model="gnb")

def test_export_pipeline_scaler():
    """Assert that a scaler is included in the pipeline."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run(["GNB", "LGB"])
    assert not isinstance(atom.export_pipeline("GNB")[0], Scaler)
    assert isinstance(atom.export_pipeline("LGB")[0], Scaler)

@patch("tempfile.gettempdir")
def test_export_pipeline_memory(func):
    """Assert that memory is True triggers a temp dir."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.scale()
    atom.export_pipeline(memory=True)
    func.assert_called_once()

@patch("pandas_profiling.ProfileReport")
def test_report(cls):
    """Assert that the report method and file are created."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.report(filename="report")
    cls.return_value.to_file.assert_called_once_with("report.html")

def test_reset():
    """Assert that the reset method deletes models and branches."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    atom.scale()
    atom.branch = "2"
    atom.encode()
    atom.run("LR")
    atom.reset()
    assert not atom.models and len(atom._branches) == 1
    assert atom["feature_3"].dtype.name == "object"  # Is reset back to str

def test_save_data():
    """Assert that the dataset is saved to a csv file."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.save_data(FILE_DIR + "auto")
    assert glob.glob(FILE_DIR + "ATOMClassifier_dataset.csv")

def test_shrink_dtypes_excluded():
    """Assert that some dtypes are excluded from changing."""
    atom = ATOMClassifier(X10_str2, y10, random_state=1)
    assert atom.dtypes[3].name == "bool"
    atom.shrink()
    assert atom.dtypes[3].name == "bool"

def test_shrink_obj2cat():
    """Assert that the obj2cat parameter works as intended."""
    atom = ATOMClassifier(X10_str2, y10, random_state=1)
    atom.shrink(obj2cat=False)
    assert atom.dtypes[2].name == "object"

    atom.shrink()
    assert atom.dtypes[2].name == "category"

def test_shrink_int2uint():
    """Assert that the int2uint parameter works as intended."""
    atom = ATOMClassifier(X10_str2, y10, random_state=1)
    assert atom.dtypes[0].name == "int64"
    atom.shrink()
    assert atom.dtypes[0].name == "int8"

    assert atom.dtypes[0].name == "int8"
    atom.shrink(int2uint=True)
    assert atom.dtypes[0].name == "uint8"

def test_shrink_sparse_arrays():
    """Assert that sparse arrays are also transformed."""
    atom = ATOMClassifier(X_sparse, y10, random_state=1)
    assert atom.dtypes[0].name == "Sparse[int64, 0]"
    atom.shrink()
    assert atom.dtypes[0].name == "Sparse[int8, 0]"

def test_shrink_dtypes_unchanged():
    """Assert that optimal dtypes are left unchanged."""
    atom = ATOMClassifier(X_bin.astype("float32"), y_bin, random_state=1)
    assert atom.dtypes[3].name == "float32"
    atom.shrink()
    assert atom.dtypes[3].name == "float32"

def test_shrink_dense2sparse():
    """Assert that the dataset can be converted to sparse."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    assert atom.dtypes[0].name == "float64"
    atom.shrink(dense2sparse=True)
    assert atom.dtypes[0].name.startswith("Sparse[float32")

def test_shrink_exclude_columns():
    """Assert that columns can be excluded."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    assert atom.dtypes[0].name == "float64"
    assert atom.dtypes[-1].name != "int8"
    atom.shrink(columns=-1)
    assert atom.dtypes[0].name == "float64"
    assert atom.dtypes[-1].name == "int8"

def test_stats_mixed_sparse_dense():
    """Assert that stats show new information for mixed datasets."""
    atom = ATOMClassifier(X_sparse, y10, random_state=1)
    atom.apply(lambda x: 1, columns="new_column")
    atom.stats()

def test_status():
    """Assert that the status method prints an overview of the instance."""
    atom = ATOMClassifier(X_class, y_class, random_state=1)
    atom.status()

def test_transform_method():
    """ Assert that the transform method works as intended."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    atom.encode(max_onehot=None)
    assert atom.transform(X10_str)["feature_3"].dtype.kind in "ifu"

def test_transform_not_train_only():
    """Assert that train_only transformers are not used."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.prune(max_sigma=2)
    assert len(atom.transform(X_bin)) == len(X_bin)

def test_transform_verbose_invalid():
    """Assert an error is raised for an invalid value of verbose."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.clean()
    pytest.raises(ValueError, atom.transform, X_bin, verbose=3)

def test_transform_with_y():
    """Assert that the transform method works when y is provided."""
    atom = ATOMClassifier(X10_nan, y10, random_state=1)
    atom.impute(strat_num="drop")
    X, y = atom.transform(X10_nan, y10)
    assert len(y) < len(y10)

def test_transform_skip_y():
    """Assert that only transformers are skipped when not provided."""
    atom = ATOMClassifier(X10_str, y10_str, random_state=1)
    atom.add(LabelEncoder())
    atom.encode()
    assert isinstance(atom.transform(X10_str), pd.DataFrame)

# Test base transformers =========================================== >>

def test_custom_params_to_method():
    """Assert that a custom parameter is passed to the method."""
    atom = ATOMClassifier(X_bin, y_bin, verbose=1, random_state=1)
    atom.scale(verbose=2)
    assert atom.pipeline[0].verbose == 2

def test_add_depending_models():
    """Assert that an error is raised when the branch has dependent models."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("LR")
    with pytest.raises(PermissionError, match=r".*allowed to add transformers.*"):
        atom.clean()

def test_add_no_transformer():
    """Assert that an error is raised if the estimator has no estimator."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    with pytest.raises(AttributeError, match=r".*should have a transform method.*"):
        atom.add(RandomForestClassifier())

def test_add_basetransformer_params_are_attached():
    """Assert that the n_jobs and random_state params from atom are used."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.add(PCA())  # When left to default
    atom.add(PCA(random_state=2))  # When changed
    assert atom.pipeline[0].get_params()["random_state"] == 1
    assert atom.pipeline[1].get_params()["random_state"] == 2

def test_add_train_only():
    """Assert that atom accepts transformers for the train set only."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.add(StandardScaler(), train_only=True)
    assert check_scaling(atom.X_train) and not check_scaling(atom.X_test)

    len_train, len_test = len(atom.train), len(atom.test)
    atom.add(Pruner(), train_only=True)
    assert len(atom.train) != len_train and len(atom.test) == len_test

def test_add_complete_dataset():
    """Assert that atom accepts transformers for the complete dataset."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.add(StandardScaler())
    assert check_scaling(atom.dataset)

    len_dataset = len(atom.dataset)
    atom.add(Pruner())
    assert len(atom.dataset) != len_dataset

def test_add_transformer_only_y():
    """Assert that atom accepts transformers with only y."""
    atom = ATOMClassifier(X10, y10_str, random_state=1)
    atom.add(LabelEncoder())
    assert np.all((atom["target"] == 0) | (atom["target"] == 1))

def test_returned_column_already_exists():
    """Assert that an error is raised if an existing column is returned."""
    atom = ATOMClassifier(X_text, y10, random_state=1)
    atom.apply(lambda x: 1, columns="new")
    with pytest.raises(RuntimeError, match=r".*already exists in the original.*"):
        atom.vectorize(columns="corpus")

def test_add_sparse_matrices():
    """Assert that transformers that return sp.matrix are accepted."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    atom.add(OneHotEncoder(handle_unknown="ignore"), columns=2)
    assert atom.shape == (10, 8)  # Creates 4 extra columns

def test_add_keep_column_names():
    """Assert that the column names are kept after transforming."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)

    # When the columns are only transformed
    atom.add(StandardScaler())
    assert atom.features.tolist() == list(X_bin)

    # When columns were removed
    atom.add(SelectFromModel(RandomForestClassifier()))
    assert all(col in list(X_bin) for col in atom.features)

def test_raise_length_mismatch():
    """Assert that an error is raised when there's a mismatch in row length."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    with pytest.raises(ValueError, match=r".*does not match length.*"):
        atom.prune(columns=[2, 4])

def test_add_derivative_columns_keep_position():
    """Assert that derivative columns go after the original."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    atom.encode(columns="feature_3")
    assert list(atom.columns[2:5]) == ["feature_3_a", "feature_3_b", "feature_3_c"]

def test_add_sets_are_kept_equal():
    """Assert that the train and test sets always keep the same rows."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    len_train, len_test = len(atom.train), len(atom.test)
    atom.add(Pruner())
    assert len(atom.train) < len_train and len(atom.test) < len_test

def test_add_reset_index():
    """Assert that the indices are reset when index=False."""
    atom = ATOMClassifier(X_bin, y_bin, index=False, random_state=1)
    atom.prune()
    assert list(atom.dataset.index) == list(range(len(atom.dataset)))

def test_add_params_to_method():
    """Assert that atom's parameters are passed to the method."""
    atom = ATOMClassifier(X_bin, y_bin, verbose=1, random_state=1)
    atom.scale()
    assert atom.pipeline[0].verbose == 1

def test_add_pipeline():
    """Assert that adding a pipeline adds every individual step."""
    pipeline = Pipeline(
        steps=[
            ("scaler", StandardScaler()),
            ("sfm", SelectFromModel(RandomForestClassifier())),
        ]
    )
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.add(pipeline)
    assert isinstance(atom.pipeline[0], StandardScaler)
    assert isinstance(atom.pipeline[1], SelectFromModel)

def test_apply_not_callable():
    """Assert that an error is raised when func is not callable."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    pytest.raises(TypeError, atom.apply, func=RandomForestClassifier(), columns=0)

def test_apply_same_column():
    """Assert that apply can transform an existing column."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.apply(lambda x: 1, columns=0)
    assert atom["mean radius"].sum() == atom.shape[0]
    assert str(atom.pipeline[0]).startswith("FuncTransformer(func=<lambda>")

def test_apply_new_column():
    """Assert that apply can create a new column."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.apply(lambda x: 1, columns="new column")
    assert atom["new column"].sum() == atom.shape[0]

def test_apply_args_and_kwargs():
    """Assert that args and kwargs are passed to the function."""

    def test_func(df, arg_1, arg_2="mean radius"):
        return df[arg_2] + arg_1

    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.apply(test_func, columns="new column", args=(10,), arg_2="mean texture")
    assert atom["new column"][0] == atom["mean texture"][0] + 10

def test_drop():
    """Assert that columns can be dropped through the pipeline."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.drop([0, 1])
    assert atom.n_features == X_bin.shape[1] - 2
    assert str(atom.pipeline[0]).startswith("DropTransformer(columns")

# Test data cleaning transformers =================================== >>

def test_scale():
    """Assert that the scale method normalizes the features."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.scale()
    assert check_scaling(atom.dataset)
    assert hasattr(atom, "standard")

def test_gauss():
    """Assert that the gauss method transforms the features."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    X = atom.X.copy()
    atom.gauss()
    assert not atom.X.equals(X)
    assert hasattr(atom, "yeojohnson")

def test_clean():
    """Assert that the clean method cleans the dataset."""
    atom = ATOMClassifier(X10, y10_sn, stratify=False, random_state=1)
    atom.clean()
    assert len(atom.dataset) == 9
    assert atom.mapping == {"target": {"n": 0, "y": 1}}

def test_impute():
    """Assert that the impute method imputes all missing values."""
    atom = ATOMClassifier(X10_nan, y10, random_state=1)
    atom.impute()
    assert atom.dataset.isna().sum().sum() == 0

def test_discretize():
    """Assert that the discretize method bins the numerical columns."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.discretize()
    assert all(dtype.name == "object" for dtype in atom.X.dtypes)

def test_encode():
    """Assert that the encode method encodes all categorical columns."""
    atom = ATOMClassifier(X10_str, y10, random_state=1)
    atom.encode()
    assert all(atom.X[col].dtype.kind in "ifu" for col in atom.X.columns)

def test_prune():
    """Assert that the prune method handles outliers in the training set."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    len_train, len_test = len(atom.train), len(atom.test)
    atom.prune(strategy="lof")
    assert len(atom.train) != len_train and len(atom.test) == len_test
    assert hasattr(atom, "lof")

def test_balance_wrong_task():
    """Assert that an error is raised for regression tasks."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    pytest.raises(PermissionError, atom.balance, oversample=0.7)

def test_balance():
    """Assert that the balance method balances the training set."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    length = (atom.y_train == 1).sum()
    atom.balance(strategy="NearMiss")
    assert (atom.y_train == 1).sum() != length
    assert hasattr(atom, "nearmiss")

# Test nlp transformers ============================================ >>

def test_textclean():
    """Assert that the textclean method cleans the corpus."""
    atom = ATOMClassifier(X_text, y10, shuffle=False, random_state=1)
    atom.textclean()
    assert atom["corpus"][0] == "i am in new york"
    assert hasattr(atom, "drops")

def test_tokenize():
    """Assert that the tokenize method tokenizes the corpus."""
    atom = ATOMClassifier(X_text, y10, shuffle=False, random_state=1)
    atom.tokenize()
    assert atom["corpus"][0] == ["I", "àm", "in", "ne", "'", "w", "york"]

def test_normalize():
    """Assert that the normalize method normalizes the corpus."""
    atom = ATOMClassifier(X_text, y10, shuffle=False, random_state=1)
    atom.normalize(stopwords=False, custom_stopwords=["yes"])
    assert atom["corpus"][0] == ["I", "àm", "in", "ne'w", "york"]

def test_vectorize():
    """Assert that the vectorize method converts the corpus to numerical."""
    atom = ATOMClassifier(X_text, y10, test_size=0.25, random_state=1)
    atom.vectorize(strategy="hashing", n_features=5)
    assert "corpus" not in atom
    assert atom.shape == (10, 6)
    assert hasattr(atom, "hashing")

# Test feature engineering transformers ============================ >>

def test_feature_extraction():
    """Assert that the feature_extraction method creates datetime features."""
    atom = ATOMClassifier(X10_dt, y10, verbose=2, random_state=1)
    atom.feature_extraction(fmt="%d/%m/%Y")
    assert atom.X.shape[1] == 6

def test_feature_generation():
    """Assert that the feature_generation method creates extra features."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.feature_generation(n_features=2)
    assert atom.X.shape[1] == X_bin.shape[1] + 2

def test_feature_generation_attributes():
    """Assert that the attrs from feature_generation are passed to atom."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.feature_generation("gfg", n_features=2)
    assert hasattr(atom, "gfg")
    assert hasattr(atom, "genetic_features")

def test_feature_selection_attrs():
    """Assert that feature_selection attaches only used attributes."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.feature_selection(strategy="univariate", n_features=8)
    assert hasattr(atom, "univariate")
    assert not hasattr(atom, "RFE")

def test_default_solver_univariate():
    """Assert that the default solver is selected for strategy="univariate"."""
    # For classification tasks
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.feature_selection(strategy="univariate", solver=None, n_features=8)
    assert atom.pipeline[0].univariate.score_func.__name__ == "f_classif"

    # For regression tasks
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.feature_selection(strategy="univariate", solver=None, n_features=8)
    assert atom.pipeline[0].univariate.score_func.__name__ == "f_regression"

def test_winner_solver_after_run():
    """Assert that the solver is the winning model after run."""
    atom = ATOMClassifier(X_class, y_class, random_state=1)
    atom.run("LR")
    atom.branch = "fs_branch"
    atom.feature_selection(strategy="sfm", solver=None, n_features=8)
    assert atom.pipeline[0].sfm.estimator_ is atom.winner.estimator

def test_default_solver_from_task():
    """Assert that the solver is inferred from the task when a model is selected."""
    # For classification tasks
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.feature_selection(strategy="rfe", solver="lgb", n_features=8)
    assert atom.pipeline[0].rfe.estimator_.__class__.__name__ == "LGBMClassifier"

    # For regression tasks
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.feature_selection(strategy="rfe", solver="lgb", n_features=25)
    assert atom.pipeline[0].rfe.estimator_.__class__.__name__ == "LGBMRegressor"

@patch("atom.feature_engineering.SequentialFeatureSelector")
def test_default_scoring(cls):
    """Assert that the scoring is atom's metric when exists."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("lr", metric="recall")
    atom.branch = "fs_branch"
    atom.feature_selection(strategy="sfs", solver="lgb", n_features=25)
    assert atom.pipeline[0].kwargs["scoring"].name == "recall"

# Test training methods ============================================ >>

def test_assign_existing_metric():
    """Assert that the existing metric_ is assigned if rerun."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("LR", metric="recall")
    atom.run("Tree")
    assert atom.metric == "recall"

def test_raises_invalid_metric_consecutive_runs():
    """Assert that an error is raised for a different metric."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run("LR", metric="recall")
    pytest.raises(ValueError, atom.run, "Tree", metric="f1")

def test_scaling_is_passed():
    """Assert that the scaling is passed to the trainer."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.scale("minmax")
    atom.run("LGB")
    assert atom.dataset.equals(atom.lgb.dataset)

def test_errors_are_updated():
    """Assert that the found exceptions are updated in the errors attribute."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)

    # Produce an error on one model (when n_initial_points > n_calls)
    atom.run(["Tree", "LGB"], n_calls=(3, 2), n_initial_points=(2, 5))
    assert list(atom.errors) == ["LGB"]

    # Subsequent runs should remove the original model
    atom.run(["Tree", "LGB"], n_calls=(5, 3), n_initial_points=(7, 1))
    assert atom.models == "LGB"

def test_models_and_metric_are_updated():
    """Assert that the models and metric attributes are updated correctly."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.run(["OLS", "Tree"], metric=get_scorer("max_error"))
    assert atom.models == ["OLS", "Tree"]
    assert atom.metric == "max_error"

def test_errors_are_removed():
    """Assert that the errors are removed if subsequent runs are successful."""
    atom = ATOMClassifier(X_bin, y_bin, random_state=1)
    atom.run(["BNB", "Tree"], n_initial_points=(2, 7))  # Invalid value for Tree
    atom.run("Tree")  # Runs correctly
    assert not atom.errors  # Errors should be empty

def test_trainer_becomes_atom():
    """Assert that the parent trainer is converted to atom."""
    atom = ATOMRegressor(X_reg, y_reg, random_state=1)
    atom.run("Tree")
    assert atom is atom.tree.T

max_stars_repo_name: tvdboom/AutoML

max_issues_repo_name: tvdboom/AutoML

max_forks_repo_name: tvdboom/AutoML


text:
import glob
import os
import numpy as np
import seaborn as sns
from keras.models import Sequential
from keras.layers import Dense,Dropout,Flatten,Conv2D,MaxPooling2D
from keras.preprocessing.image import img_to_array,array_to_img,load_img
from PIL import Image
from sklearn.preprocessing import LabelEncoder,OneHotEncoder
from sklearn.model_selection import train_test_split

imgs = glob.glob(r'C:\Users\ekrem\OneDrive\Masaüstü\ML Projects\T-rex Game\imgs\*.png')

w = 125
h = 50

X = []
y = []

filename = os.path.basename(imgs[0])
label = filename.split('_')[0]
print(filename)

for img in imgs:
    filename = os.path.basename(img)
    label = filename.split('_')[0]
    im = Image.open(img).convert('L')
    im = np.array(im)
    im = np.resize(im,(w,h))
    im = im/255
    X.append(im)
    y.append(label)

X = np.array(X)
X = X.reshape((X.shape[0],X.shape[1],X.shape[2],1))

print(y)
sns.countplot(y)

def one_hot_labels (values):
    le = LabelEncoder()
    label_encoded = le.fit_transform(values)
    onehot_encoder = OneHotEncoder(sparse=False)
    label_encoded = label_encoded.reshape((len(label_encoded),1))
    onehot_encoded = onehot_encoder.fit_transform(label_encoded)
    return onehot_encoded

y=one_hot_labels(y)
print(y[0:5])

X_train,X_test,y_train,y_test = train_test_split(X,y,test_size=0.25,
                                                 random_state=2)
print("""
      Shape of X Train:{}
      Shape of y Train:{}
      Shape of X Test :{}
      Shape of y Test :{}
      """.format(X_train.shape,y_train.shape,X_test.shape,y_test.shape))

model = Sequential()

model.add(Conv2D(32,kernel_size=(3,3),activation='relu',input_shape=(w,h,1)))
model.add(Conv2D(64,kernel_size=(3,3),activation='relu'))
model.add(MaxPooling2D(pool_size=(2,2)))
model.add(Dropout(0.25))
model.add(Flatten())
model.add(Dense(128,activation='relu'))
model.add(Dropout(0.3))
model.add(Dense(3,activation='softmax'))

model.compile(loss='categorical_crossentropy',
              optimizer='Adam',
              metrics=['accuracy'])

model.fit(X_train,y_train,epochs=35,batch_size=64)

score_train = model.evaluate(X_train,y_train)
print('Training Accuracy :'+str(score_train[1]*100))

score_test = model.evaluate(X_test,y_test)
print('Test Accuracy :'+str(score_test[1]*100))

model.save('C:\\Users\\ekrem\\OneDrive\\Masaüstü\\ML Projects\\T-rex Game\\trex_weight.h5')

max_stars_repo_name: ekrrems/T-Rex-Dino-Game-Automation

max_issues_repo_name: ekrrems/T-Rex-Dino-Game-Automation

max_forks_repo_name: ekrrems/T-Rex-Dino-Game-Automation


text:
import numpy as np

from keras.utils import Sequence

def create_center_mask(patch_size, center_size):
    mask = np.zeros(patch_size[:2])
    h, w = patch_size[:2]
    y_start = h // 2 - center_size[0] // 2
    x_start = w // 2 - center_size[1] // 2
    mask[y_start:y_start + center_size[0], x_start:x_start + center_size[1]] = 1

    return np.expand_dims(mask, axis=2)

def create_rnd_patch(img, patch_size, seed):
    radius = img.shape[0] / 2

    np.random.seed(seed)
    angle = np.random.randint(0, 359)
    distance = np.random.randint(0, radius - np.sqrt(np.sum(np.power(patch_size, 2))))
    x_start, y_start = (int(radius + distance * np.cos(angle)), int(radius - distance * np.sin(angle)))

    return img[y_start:y_start + patch_size[0], x_start:x_start + patch_size[1]]

def prepare_patch(img, patch_size, mask, seed):
    patch = create_rnd_patch(img, patch_size, seed)
    patch = (1 - mask) * patch

    return patch.astype('float32') / 255.0

class DataGenerator(Sequence):
    """Generates data for Keras"""
    def __init__(self, imgs, patch_size, center_size, batch_size=32, shuffle=True):

        self.imgs = imgs
        self.patch_size = patch_size
        self.center_size = center_size
        self.batch_size = batch_size
        self.shuffle = shuffle

        self.epoch = 0
        self.on_epoch_end()

        self.mask = create_center_mask(patch_size, center_size)

    def __len__(self):
        """Denotes the number of batches per epoch"""
        return int(np.floor(len(self.imgs) / self.batch_size))

    def __getitem__(self, index):
        """Generate one batch of data"""
        # Generate indexes of the batch
        indexes = self.indexes[index * self.batch_size:(index + 1) * self.batch_size]

        # Generate data
        x = self.__data_generation(indexes, [self.imgs[k] for k in indexes])

        return x, x

    def on_epoch_end(self):
        """Updates indexes after each epoch"""
        self.epoch += 1

        self.indexes = np.arange(len(self.imgs))
        if self.shuffle == True:
            np.random.seed(self.epoch)
            np.random.shuffle(self.indexes)

    def __data_generation(self, indexes, batch):
        """Augments or/and pretransforms data"""
        patches = [prepare_patch(img, self.patch_size, self.mask, (self.epoch * idx) % (2**32 - 1))
                   for idx, img in zip(indexes, batch)]

        return np.array(patches)


text:
from vienna_interface import parse_structured_score, parse_score
from rRNA_23S import seq_23S, struct_23S
from rRNA_16S import seq_16S, struct_16S
import numpy as np
import json
import os
import seaborn as sns
from tqdm import tqdm
import pandas as pd

# ensure uppercase
seq_16S = seq_16S.upper()
seq_23S = seq_23S.upper()

# Look, this is obviously a terrible, bizarre way to coerce a json-serialized
# dictionary into a dataframe. I cannot honestly imagine why I did this a
# couple years ago. It does work, though.
def load_cache(jsonfile):
    with open(jsonfile) as f:
        x,y = [], []
        cache = json.loads(f.read())
        for k,v in cache.items():
            y.append(v)
            x.append(k)
        return x, y

x,y = load_cache('cache_bp_23S.json')
df_23S_mc = pd.DataFrame.from_dict([{'seq': k, 'delta': v} for k,v in zip(x,y)])
x,y = load_cache('cache_bp_16S.json')
df_16S_mc = pd.DataFrame.from_dict([{'seq': k, 'delta': v} for k,v in zip(x,y)])

def count_muts(seq, wt):
    return len([0 for c, c_ in zip(seq, wt) if c != c_])

df_16S_mc['muts'] = df_16S_mc.apply(lambda x: count_muts(x.seq.upper(), seq_16S.upper()), axis=1)
df_23S_mc['muts'] = df_23S_mc.apply(lambda x: count_muts(x.seq.upper(), seq_23S.upper()), axis=1)

df = pd.read_csv('pilot_sequences.csv')
df['muts'] = df.apply(lambda x: count_muts(x.Sequence.upper(), seq_23S.upper()) if x.Subunit == '23S' else count_muts(x.Sequence, seq_16S.upper()), axis=1)

# produce sequences with these mutation numbers
print('delta;seq;muts')
for mut_num in df[(df.Round == 'Pilot') & (df.Subunit == '23S')].muts.unique():
    num = len(df[(df.Round == 'Pilot') & (df.Subunit == '23S') & (df.muts == mut_num)])
    # emit one sequence per sequence from the pilot round with that number mutations
    for ii in range(num):
        try:
            print('{};{};{}'.format(
                df_23S_mc[df_23S_mc.muts == mut_num].nsmallest(num, 'delta').delta.values[ii],
                df_23S_mc[df_23S_mc.muts == mut_num].nsmallest(num, 'delta').seq.values[ii],
                df_23S_mc[df_23S_mc.muts == mut_num].nsmallest(num, 'delta').muts.values[ii]
            ))
        except:
            print("stumped by mut_num", mut_num)

for mut_num in df[(df.Round == 'Pilot') & (df.Subunit == '16S')].muts.unique():
    num = len(df[(df.Round == 'Pilot') & (df.Subunit == '16S') & (df.muts == mut_num)])
    for ii in range(num):
        print('{};{};{}'.format(
            df_16S_mc[df_16S_mc.muts == mut_num].nsmallest(num, 'delta').delta.values[ii],
            df_16S_mc[df_16S_mc.muts == mut_num].nsmallest(num, 'delta').seq.values[ii],
            df_16S_mc[df_16S_mc.muts == mut_num].nsmallest(num, 'delta').muts.values[ii]
        ))


text:
import numpy as np

# General meteorological constants
k = 0.41  # Von Karman's constant
R_d = 287.0  # Gas constant for dry air [J kg^-1 K^-1]
R_v = 461.51  # Gas constant for water vapor [J kg^-1 K^-1]
eps = 0.622  # R_d/R_v []
c_pd = 1004.67  # Specific heat capacity of dry air at constant pressure [J kg^-1 K^-1]
g = 9.80665  # Standard Gravity (3rd CGPM) [m s^-2]
R_g = 8.3144598  # Universal gas constant [J mol^-1 K^-1]

# Heat flux constants
c_dry = 840.  # specific heat capacity (bulk value) of dry soil components [J kg^-1 K^-1]
c_wet = 4181.3  # specific heat capacity of water [J kg^-1 K^-1]
rho_w = 1000.  # bulk density of water [kg m^-3]

# Approximation for C_p and l_v terms
l_v_const = 1.45  # Latent heat of vaporization at 20°C
c_p_const = 1.013e-3  # Specific heat of air at constant pressure [MJ kg^-1 °C^-1]

# Coefficients used in calculating saturation vapor pressure (met_utils)
gs = np.array((-2.8365744e3,
              -6.028076559e3,
              1.954263612e1,
              -2.737830188e-2,
              1.6261698e-5,
              7.0229056e-10,
              -1.8680009e-13,
              2.7150305)).reshape((1,1,8))

max_stars_repo_name: miksch/topo-flux

max_issues_repo_name: miksch/topoFlux

max_forks_repo_name: miksch/topoFlux


text:
# Copyright 2022 DeepMind Technologies Limited
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Tests for open_spiel.python.algorithms.tabular_qlearner."""

from absl.testing import absltest
import numpy as np

from open_spiel.python import rl_environment
from open_spiel.python.algorithms import tabular_qlearner
import pyspiel

# Fixed seed to make test non stochastic.
SEED = 10000

# A simple two-action game encoded as an EFG game. Going left gets -1, going
# right gets a +1.
SIMPLE_EFG_DATA = """
  EFG 2 R "Simple single-agent problem" { "Player 1" } ""
  p "ROOT" 1 1 "ROOT" { "L" "R" } 0
    t "L" 1 "Outcome L" { -1.0 }
    t "R" 2 "Outcome R" { 1.0 }
"""

class QlearnerTest(absltest.TestCase):

  def test_simple_game(self):
    game = pyspiel.load_efg_game(SIMPLE_EFG_DATA)
    env = rl_environment.Environment(game=game)

    agent = tabular_qlearner.QLearner(0, game.num_distinct_actions())
    total_reward = 0

    for _ in range(100):
      total_eval_reward = 0
      for _ in range(1000):
        time_step = env.reset()
        while not time_step.last():
          agent_output = agent.step(time_step)
          time_step = env.step([agent_output.action])
          total_reward += time_step.rewards[0]
        agent.step(time_step)
      self.assertGreaterEqual(total_reward, 75)
      for _ in range(1000):
        time_step = env.reset()
        while not time_step.last():
          agent_output = agent.step(time_step, is_evaluation=True)
          time_step = env.step([agent_output.action])
          total_eval_reward += time_step.rewards[0]
      self.assertGreaterEqual(total_eval_reward, 250)

if __name__ == "__main__":
  np.random.seed(SEED)
  absltest.main()


text:
from binance.client import Client, BinanceAPIException
from binance.enums import *
import numpy as np
import math
import time as t
from datetime import datetime, timezone

decimal_places = 4
min_buy = 55.00
max_buy = 60.00

def calculate_ma50(symbolTicker, client):
    ma50_local = 0
    sum = 0

    klines = client.get_historical_klines(symbolTicker, Client.KLINE_INTERVAL_15MINUTE, "15 hour ago UTC")

    if (len(klines) == 60):

        for i in range(10,60):
            sum = sum + float(klines[i][4])

        ma50_local = sum / 50

    return ma50_local

def orderStatus(orderToCkeck, symbolTicker):
    try:
        status = client.get_order(
            symbol = symbolTicker,
            orderId = orderToCkeck.get('orderId')
        )
        return status.get('status')
    except Exception as e:
        print(e)
        return 7

def tendencia_ma50_4hs_15minCandles(symbolTicker, client):
    x = []
    y = []
    sum = 0
    ma50_i = 0

    t.sleep(1)

    resp = False

    klines = client.get_historical_klines(symbolTicker, Client.KLINE_INTERVAL_15MINUTE, "18 hour ago UTC")

    if (len(klines) != 72):
        return False
    for i in range(56,72):
        for j in range(i-50,i):
            sum = sum + float(klines[j][4])
        ma50_i = round(sum / 50,8)
        sum = 0
        x.append(i)
        y.append(float(ma50_i))

    modelo = np.polyfit(x, y, 1)

    if (modelo[0]>0):
        resp = True

    return resp

def calculate_quantity_buy(crypto, client, min_buy, max_buy):
    """
    This function receives a symbol as a parameter, and returns the minimum amount to 10.00 USDT

    :params crypto:

    :type crypto: str

    returns the minimum quantity to Buy
    """
    current_price = client.get_symbol_ticker(symbol=crypto)
    get_price = float(current_price['price'])

    if get_price >= min_buy and get_price <= max_buy:
        print(f'Current price bigger then 10, price {get_price} USDT | Buy quantity 1')
        return 1
    elif get_price <= min_buy:
        quantity = 1
        while True:
            quantity_current = get_price * quantity
            if quantity_current < min_buy:
                quantity +=1
                continue
            break
        print(f'Current price less than 10, price unit {get_price} -- price buy {get_price*quantity} USDT | Buy quantity {quantity}')
        return quantity

def format_Price_decimal_percente(priceToFormat, percente, decimal) -> float:
    """
    This function returns a price formatted to 8 decimal places

    :params priceToFormat:
    :params percente:

    :type priceToFormat: str
    :type percente: float

    returns a formatted price, already with a percentage added
    """
    price = float(priceToFormat)
    result = price*percente
    return f"{round(float(result),decimal)}"

def check_balance(symbol, client):
    """
    This function returns a Boolean value

    :params symbol:

    :type symbol: str

    returns a bool, True if there is a symbol in the wallet and False if there is no symbol
    """
    balance = client.get_asset_balance(asset=symbol)
    get_quantity = float(balance['free'])
    if get_quantity == 0.0:
        return {"status": False, "quantity": get_quantity}
    return {"status": True, "quantity": get_quantity}

def buy_stop_loss_limit(client, symbolTicker, quantityBuy, priceBuy, stopPriceBuy):
    """
    This function returns a buy order with stop limit

    :params client:
    :params symbolTicker:
    :params quantityBuy:
    :params priceBuy:
    :params stopPriceBuy:

    :type client: Binance client instance
    :type symbolTicker: str
    :type quantityBuy: int
    :type priceBuy: str
    :type stopPriceBuy: str

    returns a dict with orderId to buy
    """
    buyOrder = client.create_order(
        symbol=symbolTicker,
        side='BUY',
        type='STOP_LOSS_LIMIT',
        quantity=quantityBuy,
        price=priceBuy,
        stopPrice=stopPriceBuy,
        timeInForce='GTC')
    return buyOrder

def sell_order_OCO(client, symbolTicker, quantitySell, priceSell, stopPriceSell, stopLimitPriceSell):
    """
    This function returns a sell order with stop limit

    :params client:
    :params symbolTicker:
    :params quantitySell:
    :params priceSell:
    :params stopPriceSell:
    :params stopLimitPriceSell:

    :type client: Binance client instance
    :type symbolTicker: str
    :type quantitySell: int
    :type priceSell: str
    :type stopPriceSell: str
    :type stopLimitPriceSell: str

    returns a dict with orderId to sell
    """

    orderOCO = client.order_oco_sell(
        symbol = symbolTicker,
        quantity = quantitySell,
        price = priceSell,
        stopPrice = stopPriceSell,
        stopLimitPrice =stopLimitPriceSell,
        stopLimitTimeInForce = 'GTC')
    return orderOCO

def get_price_ticker(symbol, client):
  """
  This function returns the price of the symbol

  :params symbol:
  :params client:

  :type symbol: str
  :type client: Binance instance client

  return a price in float
  """
  list_of_tickers = client.get_all_tickers()
  for tick in list_of_tickers:
    if tick['symbol'] == symbol:
      symbolPrice = float(tick['price'])
      break
  return symbolPrice

def show_updated_prices(symbolTicker,ma50, symbolPrice):
    """
    This function displays prices

    :params symbolTicker:
    :params ma50:
    :params symbolPrice:

    :type symbolTicker: str
    :type ma50: float
    :type symbolPrice: str
    """
    print("********** " + symbolTicker + " **********")
    print(" ActualMA50: "  + str(round(ma50,decimal_places)))
    print("ActualPrice: " + str(round(symbolPrice,decimal_places)))
    print(" PriceToBuy: "  + str(round(ma50*0.99,decimal_places)))
    print("----------------------")

def formatForPriceDecimal(priceToFormat: str, decimal: float) -> float:
    return  f"{round(float(priceToFormat), decimal)}"

def signal_for_sell(percenteSell: float, list_all_tickers: list, symbolTicker: str, priceCompare: float) -> bool:
    """ This function compares the price that was purchased with the current price of the asset

    Args:
        percenteSell (float): Percentage of sale
        list_all_tickers (list): List with all assets
        symbolTicker (str): Tick base
        priceCompare (float): price that was bought active

    Returns:
        bool: Boolean return signaling sale
    """
    result = list(filter(lambda tick : (tick['symbol'] == symbolTicker), list_all_tickers))
    price_compare_buy = float(format_Price_decimal_percente(priceCompare, percenteSell, decimal_places))
    if price_compare_buy < float(result[0]['price']):
        return True
    return False

def get_price_current(list_all_tickers: list, symbolTicker: str) -> float:
    """This function returns the current price of an asset

    Args:
        list_all_tickers (list): List with all assets
        symbolTicker (str): Tick base

    Returns:
        float: returns the current price
    """

    result = list(filter(lambda tick : (tick['symbol'] == symbolTicker), list_all_tickers))
    return round(float(result[0]['price']), decimal_places)

def Dinamic_Buy(symbolTicker: str, symbolBase: str, client: object, percentePriceBUY: float, percentePriceStopBUY: float ) -> float:
    """This function returns a purchase value. Buying is dynamic, always trying to get the lowest price.

    Args:
        symbolTicker (str): cryptocurrency par symbol
        symbolBase (str): symbol of a single asset. EX: XRP
        client (object): Binance instance client
        percentePriceBUY (float): Percentage to purchase
        percentePriceStopBUY (float): Percentage to stop buying limit

    Returns:
        float: Returns the purchased amount
    """

    return_buy = {
        "amount_buy": 0.0,
        "quantity": 0,
        "order_id": 0
    }
    status = 'NEW'

    try:
        list_of_tickers = client.get_all_tickers()
        prev_symbolPrice = get_price_current(list_of_tickers, symbolTicker)
        quantityBuy = calculate_quantity_buy(symbolTicker, client, min_buy, max_buy)
        priceBuy = format_Price_decimal_percente(prev_symbolPrice, percentePriceBUY, decimal_places)
        stopPriceBuy = format_Price_decimal_percente(prev_symbolPrice, percentePriceStopBUY, decimal_places)

        quantityBuy = calculate_quantity_buy(symbolTicker, client, min_buy, max_buy)

        t.sleep(3)
        # buy order
        buyOrder = buy_stop_loss_limit(client, symbolTicker, quantityBuy, priceBuy ,stopPriceBuy)

        return_buy['amount_buy'] = priceBuy
        return_buy['order_id'] = buyOrder['orderId']

        status_list = client.get_all_orders(symbol=symbolTicker, orderId=buyOrder['orderId'])
        for get_status in status_list:
            if get_status['status'] == 'NEW':
                status = get_status['status']
                break

    except BinanceAPIException as e:
        print (e.status_code)
        print (e.message)

    while status == 'NEW':
        checke_symbol_price = check_balance(symbolBase, client)

        print('status while ' , status)

        if float(checke_symbol_price['quantity']) > 2.50:
            print(f"Balance in account {checke_symbol_price['quantity']}")
            return_buy['quantity'] = checke_symbol_price['quantity']
            break

        print(f"Balance in account {checke_symbol_price['quantity']}")
        t.sleep(3)

        list_of_tickers = client.get_all_tickers()
        current_symbolPrice = get_price_current(list_of_tickers, symbolTicker)

        print("    Prev Price = " + str(prev_symbolPrice))
        print(" Current Price = " + str(current_symbolPrice))

        if ( prev_symbolPrice > current_symbolPrice):

            try:

                result = client.cancel_order(
                    symbol = symbolTicker,
                    orderId = buyOrder['orderId']
                )

                priceBuy = format_Price_decimal_percente(current_symbolPrice, percentePriceBUY, decimal_places)
                stopPriceBuy = format_Price_decimal_percente(current_symbolPrice, percentePriceStopBUY, decimal_places)
                quantityBuy = calculate_quantity_buy(symbolTicker, client, min_buy, max_buy)

                # buy order
                buyOrder = buy_stop_loss_limit(client, symbolTicker, quantityBuy, priceBuy ,stopPriceBuy)
                return_buy['amount_buy'] = priceBuy
                return_buy['order_id'] = buyOrder['orderId']

                status_list = client.get_all_orders(symbol=symbolTicker, orderId=buyOrder['orderId'])
                for get_status in status_list:
                    if get_status['status'] == 'NEW':
                        status = get_status['status']
                        break

                prev_symbolPrice = current_symbolPrice
                print('-- ', status)
                if status == "NEW":
                    continue
            except BinanceAPIException as e:
                print(f'An exception occurred {e}')
                print(f"status {e.status_code}")
                print(f'Message {e.message}')
                continue

    print(f"{symbolTicker} purchased value {prev_symbolPrice}")
    return return_buy

def getId_data_base(list_table: list) -> int:
    """This function returns 'id' from the table

    Args:
        list_table (list): list with database query result

    Returns:
        int: Returns a valid table 'id'
    """
    id = None
    if len(list_table) > 0:
        for get_id in list_table:
            id = get_id[0]
        return id
    return id

def get_priceBuy_Quantity(list_table_report: list) -> dict:
    """This function returns a dict with the purchase price and the quantity purchased

    Args:
        list_table_report (list): List with the last buy record
        Ex: [(2, Decimal('1.5821'), 0, Decimal('0.0000'), 10, 'XRPUSDT', datetime.date(2021, 5, 2), False)]

    Returns:
        dict: Returns a dict with "price_buy" and "quantity"
    """

    dict_info = {
        "price_buy": None,
        "quantity" : None
    }
    if len(list_table_report) != 0:
        for get_info in list_table_report:
            dict_info['price_buy'] = get_info[1]
            dict_info['quantity'] = get_info[4]
        return dict_info
    return dict_info

def round_down(n, decimals=0):
    multiplier = 10 ** decimals
    return int(math.floor(n * multiplier) / multiplier)

def calculate_profit(priceBuy: float, priceSell: float) -> float:
    return round(float((priceBuy/priceSell)*100),2)

def show_await(symbol: str, priceBuy: float, current_price: float, quantity: int, percenteSell: float) -> str:

    print(f"""
    ---------- {symbol} ---------------------
    | price buy:........... {priceBuy}
    | price sell:...........{float(format_Price_decimal_percente(priceBuy, percenteSell, decimal_places))}
    | current price:....... {current_price}
    | quantity for sale:... {quantity}
    ----------------------------------------
    """)

def STARTING_SALE(symbolTicker: str, symbolBase: str, client: object, percentagePriceSELL: float, percentageStopPriceSELL: float, percentagestopLimitPriceSELL: float, dict_info: dict) -> dict:
    """[summary]

    Args:
        symbolTicker (str): [description]
        symbolBase (str): [description]
        client (object): [description]
        percentagePriceSELL (float): [description]
        percentageStopPriceSELL (float): [description]
        percentagestopLimitPriceSELL (float): [description]
        dict_info (dict): [description]

    Returns:
        dict: [description]
    """

    try:

        return_sell = {
            "priceSell": None,
            "quantity": None
        }
        list_of_tickers = client.get_all_tickers()
        symbolPriceSale = get_price_current(list_of_tickers, symbolTicker)

        priceSell = format_Price_decimal_percente(str(symbolPriceSale), percentagePriceSELL, decimal_places)
        stopPriceSell = format_Price_decimal_percente(str(symbolPriceSale), percentageStopPriceSELL, decimal_places)
        stopLimitPriceSell = format_Price_decimal_percente(str(symbolPriceSale), percentagestopLimitPriceSELL, decimal_places)

        if dict_info['price_buy'] == None and dict_info['quantity'] == None:
            print(f'Error in {dict_info}')

        quantitySell = round_down(dict_info['quantity'])

        orderOCO = sell_order_OCO(client, symbolTicker, quantitySell, priceSell, stopPriceSell, stopLimitPriceSell)

        while True:

            t.sleep(2)
            orders = client.get_open_orders(symbol=symbolTicker)
            if(len(orders) != 0):
                for key, value in orders[0].items():
                    if key == 'price':
                        return_sell['priceSell'] = value
                    if key == 'origQty':
                        return_sell["quantity"] = value
                print('Open Orders on Sale')
                t.sleep(5)
                continue
            else:
                return return_sell
                break

    except BinanceAPIException as e:
        print(e)
        with open("Error_Bot.txt", "a") as myfile:
            myfile.write(str(datetime.now().strftime("%d-%m-%y %H:%M:%S")) +" - an exception occured - {}".format(e)+ " Oops 3 ! \n")

def body_email_buy(symbolTicker: str, price_current: float, price_buy: float,percentagePriceSELL: float, quantity_buy: float) -> str:
    price = round(float(price_buy*quantity_buy), 3)
    price_sell = float(price_buy*percentagePriceSELL)
    html = f"""\

Compra efetuada com sucesso 🤖 !!

Date: {str(datetime.now().strftime("%d-%m-%y %H:%M:%S"))} ⌚

                        Asset
                        Price Current
                        Price for Sale
                        Price Buy
                        Quantity

                        {symbolTicker}
                        {price_current} USDT
                        {price_sell} USDT
                        {price} USDT
                        {quantity_buy}

            """
    return html

def body_email_sell(symbolTicker: str, price_current: float, price_buy: float,price_sell: float, profit: float, quantity_sell: float) -> str:
    price_sale = round(float(price_sell*quantity_sell), 3)
    html = f"""\

Venda efetuada com sucesso 🤖 !!

Date: {str(datetime.now().strftime("%d-%m-%y %H:%M:%S"))} ⌚

                        Asset
                        Price Current
                        Price Sale
                        Price Buy
                        Quantity
                        Profit

                        {symbolTicker}
                        {price_current} USDT
                        {price_sale} USDT
                        {price_buy} USDT
                        {quantity_sell}
                        {profit} USDT

            """
    return html

def SELL_MARKET(symbolTicker: str, quantity: int, client: object) -> dict:
    """ This function executes a market sales order

    Args:
        symbolTicker (str): Pair symbol crypto
        quantity (int): Quantity for sale
        client (object): Binance API Client Instance

    Returns:
        dict: Returns a dict with sale information
    """

    try:
        order = client.order_market_sell(symbol=symbolTicker,quantity=quantity)
        return order
    except BinanceAPIException as error:
        print(f"Error on sell market {error}")

max_stars_repo_name: fabriciocovalesci/BotTrading

max_issues_repo_name: fabriciocovalesci/BotTrading

max_forks_repo_name: fabriciocovalesci/BotTrading


text:
#!/usr/bin/env python

## A simulated plasma line spectrum is generated with python and sent to the
## wrapped CUDA function. This acts as a test and the result will be checked
## to confirm the code is working on the system. Example plotting code is left
## commented at the bottom.

import sys
import numpy as np
import matplotlib.pyplot as plt
import pl_wrapper

sr = 25 # range gate step size
def get_simulated_ipp(L=sr*10000,alt=sr*5000,freq=5e6,txlen_us=1000):
    """Simulated send and return signals. """

    txlen = txlen_us * sr
    echo = np.zeros(L,dtype=np.complex64)\
           + np.array((np.random.randn(L) + np.random.randn(L)*1j),\
                      dtype=np.complex64)

    tx = np.zeros(txlen, dtype=np.complex64)
    tx_bits = np.sign(np.random.randn(txlen_us))
    idx = np.arange(txlen_us) * sr

    for i in range(sr):
        tx[idx+i] = tx_bits

    tvec = np.arange(txlen) / (float(sr) * 1e6)

    upshifted = np.zeros(len(tvec), dtype=np.complex64)
    for i in range(50):
        upshifted += np.exp(1j * 1000.0 * np.random.randn(1))[0]\
                     * np.exp(1j * 2.0 * np.pi * (freq + np.random.randn(1)[0] * 1e3) * tvec)

    downshifted = np.zeros(len(tvec), dtype=np.complex64)
    for i in range(50):
        downshifted += np.exp(1j * 1000.0 * np.random.randn(1))[0]\
                       * np.exp(-1j * 2.0 * np.pi * (freq + np.random.randn(1)[0] * 1e3) * tvec)

    echo[alt:(alt+txlen)] = echo[alt:(alt+txlen)] + tx * (upshifted+downshifted)

    return((echo,tx))

if __name__ == "__main__":

    print "\nSimulated spectral analysis launching."

    # example parameters
    n_ipp = 100
    tx_length = 16384
    ipp_length = 250000
    n_range_gates = 4096
    range_gate_step = 25
    range_gate_start = 0
    tx = np.zeros([tx_length * n_ipp], dtype=np.complex64)
    echo = np.zeros([ipp_length * n_ipp], dtype=np.complex64)
    spectrum = np.zeros([tx_length * n_range_gates], dtype=np.float32)

    # simulate n_ipp plasma line echoes and transmit pulses
    print "Simulating data with CPU"
    for i in range(n_ipp):
        sys.stdout.write('.')
        sys.stdout.flush()
        (echo_ipp, tx_ipp) = get_simulated_ipp(L=25*10000, txlen_us=400, alt=0)
        tx[i * tx_length + np.arange(25*400)] = np.conj(tx_ipp)
        echo[i * ipp_length + np.arange(ipp_length)] = echo_ipp
    print "\nSimulation complete."

    pl_wrapper.process_echoes(tx, echo, tx_length, ipp_length, n_ipp,\
                              spectrum, n_range_gates,
                              range_gate_step, range_gate_start)

    # check the simulation works as intended
    print "Checking for accuracy:"
    spectrum.shape = (n_range_gates, tx_length)
    shifted = np.fft.fftshift(spectrum[0,:])

    maxVal1 = max(shifted)
    maxIndex1 = np.argmax(shifted)
    shifted = np.delete(shifted, np.arange(maxIndex1 - 8, maxIndex1 + 8, 1))

    maxVal2 = max(shifted)
    maxIndex2 = np.argmax(shifted)
    shifted = np.delete(shifted, np.arange(maxIndex2 - 8, maxIndex2 + 8, 1))

    count = 0
    if (1e10 < maxVal1 < 1e12) and (1e10 < maxVal2 < 1e12):
        print "Passed spectrum maxima values test."
        count += 1
    else:
        print "Failed spectrum maxima values test."

    if all(i < 1e10 for i in shifted) is True:
        print "Passed spectrum elsewhere values test."
        count += 1
    else:
        print "Failed spectrum elsewhere values test."

    if (4900 < maxIndex1 < 4930) and (11440 < maxIndex2 < 11485):
        print "Passed spectrum maxima locations test."
        count += 1
    elif (4900 < maxIndex2 < 4930) and (11440 < maxIndex1 < 11485):
        print "Passed spectrum maxima locations test."
        count += 1
    else:
        print "Failed spectrum maxima locations test."

    if count == 3:
        print "\nPassed all tests. Working as intended.\n"
    else:
        print "\nFailed at least one test. Troubleshooting needed.\n"

    ## plotting
    ### plasma line plot
    #spectrum.shape = (n_range_gates, tx_length)
    #plt.plot(np.linspace(-12.5, 12.5, num=tx_length), np.fft.fftshift(spectrum[0,:]))
    #plt.show()
    ### log color plot
    #smin = np.median(10.0 * np.log10(spectrum))
    #smax = np.max(10.0 * np.log10(spectrum))
    #plt.imshow(10.0 * np.log10(spectrum[2000:0:-1,:]), aspect = "auto",\
               #extent = [-12.5,12.5,0,1000], vmin = smin, vmax = 100,\
               #cmap = "nipy_spectral")
    ##plt.pcolormesh(spectrum)
    #plt.colorbar()
    #plt.show()

max_stars_repo_name: njhilliard/isr-plasmaline-gpu

max_issues_repo_name: njhilliard/isr-plasmaline-gpu

max_forks_repo_name: njhilliard/isr-plasmaline-gpu