text:
from numpy.core.fromnumeric import argmax
import bandits
import utils
import numpy as np
from math import sin
import matplotlib.pyplot as plt
import numpy as np
import random as rdm

HORIZON =100

# b = bandits.BernoulliBandit (np.array ([0.3, 0.42, 0.4]))

# def version_1(b,n=1000):
#     gains = []
#     regret_cumulate = []
#     res = 0
#     regret=0
#     for i in range(n):
#         arm = np.random.randint(b.nbr_arms)
#         if arm != b.best_arm:
#             regret += 1
#             regret_cumulate.append(regret)
#         res += b.pull(arm)
#         gains.append(res)
#     return gains,regret_cumulate

b1 = bandits.NormalBandit(np.array ([0.1, 0.3, 0.4, 0.75, 0.8, 0.9, 0.95]))
b2 = bandits.BernoulliBandit(np.array ([0.1, 0.3, 0.4, 0.75, 0.8, 0.9, 0.95]))

def roundRobin(b,horizon):
    regrets = []
    rewards = []

    for i in range(horizon):
        arm = i % b.nbr_arms
        # cumulative regrets
        if len(regrets)!=0:regrets.append(regrets[-1] + b.regrets[arm])
        else: regrets.append(b.regrets[arm])

        # cumulative rewards
        if len(rewards) != 0 :  rewards.append(rewards[-1] + b.pull(arm))
        else: rewards.append(b.pull(arm))
    return regrets

def round_robin(bandit, horizon):
    nbr_echantillon = np.zeros(bandit.nbr_arms)
    Regret = np.zeros(horizon)

    for t in range(horizon):
        a = utils.randamin(nbr_echantillon)
        nbr_echantillon[a] += 1
        r = bandit.regrets[a]
        Regret[t] = Regret[max(0, t-1)] + r
    return Regret

def etc(b,horizon):
    nbr_echantillon = np.zeros(b.nbr_arms)
    Gains = np.zeros(b.nbr_arms)
    Regret = np.zeros(horizon)
    t = 0
    while t < horizon:
        if t <= 300:
            a = utils.randamin(nbr_echantillon)
            nbr_echantillon[a] += 1
            r = b.regrets[a]
            g = b.pull(a)
            Regret[t] = Regret[max(0, t-1)] + r
            Gains[a] += g
        else:
            mean_arms = np.divide(Gains,nbr_echantillon)
            arm_chosen = np.argmax(mean_arms)
            r = b.regrets[arm_chosen]
            Regret[t] = Regret[max(0, t-1)] + r

        t+=1
    return Regret

def ucb(bandit,horizon):
    nbr_echantillon = np.zeros(bandit.nbr_arms)
    moyennes_empirique = np.zeros(bandit.nbr_arms)
    Regret = np.zeros(horizon)

    for t in range(horizon):
        a = utils.randamax(moyennes_empirique + np.sqrt(2 * np.log(t+1) / (nbr_echantillon + 1)))
        echantillon = bandit.pull(a)
        moyennes_empirique[a] = nbr_echantillon[a] * moyennes_empirique[a] + echantillon
        nbr_echantillon[a] += 1
        moyennes_empirique[a] = moyennes_empirique[a] / nbr_echantillon[a]
        r = bandit.regrets[a]
        Regret[t] = Regret[max(0, t-1)] + r
    return Regret

def imed(bandit,horizon):
    nbr_echantillon = np.zeros(bandit.nbr_arms)
    moyennes_empirique = np.zeros(bandit.nbr_arms)
    Regret = np.zeros(horizon)

    for t in range(horizon):
        a = utils.randamin( (nbr_echantillon ) * utils.klGaussian(moyennes_empirique,np.max(moyennes_empirique)) + np.log(nbr_echantillon ))
        echantillon = bandit.pull(a)
        moyennes_empirique[a] = nbr_echantillon[a] * moyennes_empirique[a] + echantillon
        nbr_echantillon[a] += 1
        moyennes_empirique[a] = moyennes_empirique[a] / nbr_echantillon[a]
        r = bandit.regrets[a]
        Regret[t] = Regret[max(0, t-1)] + r
    return Regret

def plusieursXP(algo, bandit, horizon, N):
    Regrets = np.zeros((N, horizon))
    for exp in range(N):
        R = algo(bandit, horizon)
        Regrets[exp] = R
    Regret_moyen = np.mean(Regrets, axis=0)
    Regret_std = np.std(Regrets, axis=0)
    return Regret_moyen, Regret_std

if __name__ == "__main__":
    np.random.seed(12345)
    normal_bandit = bandits.NormalBandit(np.array ([0.1, 0.3, 0.4, 0.75, 0.8, 0.9, 0.95]))
    bernoulli_bandit = bandits.BernoulliBandit(np.array ([0.1, 0.3, 0.4, 0.75, 0.8, 0.9, 0.95]))
    EXP = 200

    r_ucb, std_ucb = plusieursXP(ucb, normal_bandit, HORIZON, EXP)
    plt.plot(r_ucb, label="UCB")
    plt.fill_between(np.arange(HORIZON), np.maximum(0, r_ucb - std_ucb), r_ucb + std_ucb, alpha=0.3)

    r_etc, std_etc = plusieursXP(etc, normal_bandit, HORIZON, EXP)
    plt.plot(r_etc, label="ETC")
    plt.fill_between(np.arange(HORIZON), np.maximum(0, r_etc- std_etc), r_etc + std_etc, alpha=0.3)

    r_rr, std_rr = plusieursXP(round_robin, normal_bandit, HORIZON, EXP)
    plt.plot(r_rr, label="Round Robin")
    plt.fill_between(np.arange(HORIZON), np.maximum(0, r_rr - std_rr), r_rr + std_rr, alpha=0.3)

    r_imed, std_imed = plusieursXP(imed, normal_bandit, HORIZON, EXP)
    plt.plot(r_imed, label="IMED")
    plt.fill_between(np.arange(HORIZON), np.maximum(0, r_imed - std_imed), r_imed + std_imed, alpha=0.3)

    plt.title(f"Regret - Horizon = {HORIZON}, Nombre d'XPs = {EXP}")
    plt.legend()
    plt.show()

max_stars_repo_name: AbdouAkaen/Machine-Learning

max_issues_repo_name: AbdouAkaen/Machine-Learning

max_forks_repo_name: AbdouAkaen/Machine-Learning

reason: import numpy,from numpy


text:
from autograd import numpy as np
from contextlib import contextmanager # For testing stdout
import paragami
from io import StringIO
import sys # For testing stdout

# For testing stdout
# https://gist.github.com/mogproject/fc7c4e94ba505e95fa03
@contextmanager
def captured_output():
    new_out, new_err = StringIO(), StringIO()
    old_out, old_err = sys.stdout, sys.stderr
    try:
        sys.stdout, sys.stderr = new_out, new_err
        yield sys.stdout, sys.stderr
    finally:
        sys.stdout, sys.stderr = old_out, old_err

# This class will be used for testing.
class QuadraticModel(object):
    def __init__(self, dim):
        # Put lower bounds so we're testing the contraining functions
        # and so that derivatives of all orders are nonzero.
        self.dim = dim
        self.theta_pattern = \
            paragami.NumericArrayPattern(shape=(dim, ), lb=-10.)
        self.lambda_pattern = \
            paragami.NumericArrayPattern(shape=(dim, ), lb=-10.0)

        vec = np.linspace(0.1, 0.3, num=dim)
        self.matrix = np.outer(vec, vec) + np.eye(dim)

        self.lam = self.get_default_lambda()

    def get_default_lambda(self):
        return np.linspace(0.5, 10.0, num=self.dim)

    def get_hyper_par_objective(self, theta, lam):
        # Only the part of the objective that dependson the hyperparameters.
        return lam @ theta

    def get_objective(self, theta, lam):
        objective = 0.5 * theta.T @ self.matrix @ theta
        shift = self.get_hyper_par_objective(theta, lam)
        return objective + shift

    # Testing functions that use the fact that the optimum has a closed form.
    def get_true_optimal_theta(self, lam):
        return -1 * np.linalg.solve(self.matrix, lam)

ext: py

max_stars_repo_name: rgiordan/paragami

max_issues_repo_name: rgiordan/paragami

max_forks_repo_name: rgiordan/paragami

reason: import numpy


text:
"""Includes model files from workshop notebook directory."""

import os

import numpy as np
import tensorflow as tf

_DERIVED_FMT = os.path.join(os.path.dirname(__file__), '%s.py')

# convert_sparse() -- needed by make_input_fn_stroke()
execfile(_DERIVED_FMT % '4_convert_sparse', globals(), locals())
# make_input_fn_stroke()
execfile(_DERIVED_FMT % '4_input_fn_stroke', globals(), locals())

# get_nth() -- needed by get_logits_stroke()
execfile(_DERIVED_FMT % '4_get_nth', globals(), locals())
# get_logits_stroke()
execfile(_DERIVED_FMT % '4_get_logits_stroke', globals(), locals())
# make_model_fn()
execfile(_DERIVED_FMT % '4_make_model_fn', globals(), locals())

def build_estimator(model_dir, n_classes, params, config):
  model_fn = make_model_fn(get_logits_fn=get_logits_stroke, n_classes=n_classes)
  return tf.estimator.Estimator(model_fn=model_fn, model_dir=model_dir,
                                config=config)

def serving_input_fn():
  inputs = {
    'stroke': tf.placeholder(tf.float32, [None, 3, MAX_LEN]),
    'stroke_len': tf.placeholder(tf.int64, [None]),
  }
  return tf.estimator.export.ServingInputReceiver(inputs, inputs)


text:
import logging

import numpy as np

from asreview.init_sampling import sample_prior_knowledge
from asreview.review import BaseReview
from fractions import Fraction

class ReviewSubSimulate(BaseReview):
    name = "sub_simulate"

    def __init__(
            self, as_data, *args, n_prior_included=0, n_prior_excluded=0,
            prior_idx=None, seed=None, fraction_included=None,
            n_paper_sim=None,
            **kwargs):
        if as_data.labels is None:
            raise ValueError("Cannot simulate without labels!")
        r = np.random.RandomState(seed)

        # Remove unlabeled papers.
        labels = as_data.labels
        labeled_idx = np.where((labels == 0) | (labels == 1))[0]
        if len(labeled_idx) != len(labels):
            logging.warning("Simulating partial review, ignoring unlabeled"
                            f" papers (n={len(labels)-len(labeled_idx)}.")
            as_data = as_data.slice(labeled_idx)
            labels = as_data.labels

        zero_idx = np.where(labels == 0)[0]
        one_idx = np.where(labels == 1)[0]
        n_papers = len(as_data)

        if fraction_included is None:
            fraction_included = one_idx/n_papers

        n_zero_left = len(zero_idx) - n_prior_excluded
        n_one_left = len(one_idx) - n_prior_included
        if n_paper_sim is None or n_zero_left + n_one_left < n_paper_sim:
            if n_one_left/fraction_included > n_zero_left/(1-fraction_included):
                n_zero_sim = max(1, n_zero_left)
                n_one_sim = fraction_included/(1-fraction_included)*n_zero_left
                n_one_sim = max(1, n_one_sim)
                if n_one_sim + n_prior_included < 2:
                    n_one_sim = 2
            else:
                n_one_sim = max(1, n_one_left)
                n_zero_sim = (1-fraction_included)/fraction_included*n_one_left
                n_zero_sim = max(1, n_zero_sim)
                if n_zero_sim + n_prior_excluded < 2:
                    n_zero_sim = 2
        else:
            n_one_desired = round(fraction_included*n_paper_sim)
            n_zero_desired = n_paper_sim - n_one_desired
            if n_one_desired <= n_one_left and n_zero_desired <= n_zero_left:
                n_zero_sim = n_zero_desired
                n_one_sim = n_one_desired
            elif n_one_desired < n_one_left:
                n_one_sim = n_one_desired
                n_zero_sim = max(1, (1-fraction_included)/fraction_included*n_one_desired)
            else:
                n_zero_sim = n_zero_desired
                n_one_sim = max(1, fraction_included/(1-fraction_included)*n_zero_desired)
            if n_zero_sim + n_prior_excluded < 2:
                n_zero_sim = 2
            if n_one_sim + n_prior_included < 2:
                n_one_sim = 2

        if (n_one_sim > n_one_left or n_one_sim < 2 or
                n_zero_sim > n_zero_left or n_zero_sim < 2):
            raise ValueError("Cannot find valid configuration."
                             f" {n_one_sim} {n_one_left}, {n_zero_sim}"
                             f" {n_zero_left}, {fraction_included}.")

        slice_zero_idx = r.choice(zero_idx, n_zero_sim+n_prior_excluded, replace=False)
        slice_one_idx = r.choice(one_idx, n_one_sim+n_prior_included, replace=False)

        as_data = as_data.slice(np.concatenate((one_idx, zero_idx)))

        if n_prior_included + n_prior_excluded > 0:
            start_idx = sample_prior_knowledge(
                labels, n_prior_included, n_prior_excluded)
        else:
            start_idx = np.array([])

        super(ReviewSubSimulate, self).__init__(
            as_data, *args, start_idx=start_idx, **kwargs)

    def _get_labels(self, ind):
        """Get the labels directly from memory.

        Arguments
        ---------
        ind: list, np.array
            A list with indices

        Returns
        -------
        list, np.array
            The corresponding true labels for each indice.
        """

        return self.y[ind, ]

max_stars_repo_name: qubixes/asreview-pro

max_issues_repo_name: qubixes/asreview-pro

max_forks_repo_name: qubixes/asreview-pro


text:
import os
import pandas as pd
import numpy as np
import csv
import glob
import statistics as stat
from datetime import datetime
from csv import reader
import matplotlib.pyplot as plt
import seaborn as sns
from pathlib import Path, PureWindowsPath
import Functions_malawi
from itertools import zip_longest
import plotly.express as px
import plotly.graph_objects as go

Path = "C:/Users/gvros/Desktop/4N - Copy and Code Worksheet.csv"

gimie = pd.read_csv(Path, skiprows = 0)

print(gimie.iloc[0,0])
hh = gimie.iloc[:,0]
JFK_SUM_Time = []
JFK_SUM_Time = []
#for row, col in eumerate(hh):

#with open(Path, 'r') as data:
#    for line in csv.DictReader(data):
#        print(line)

# first round is for 1N and 10071007

Phase = "2N"
Houshold = 1007

Path_CE_MErge = "C:/Users/gvros/Desktop/Oregon State Masters/Work/OSU, CSC, CQC Project files/"+Phase+"/Compiler_1_exact/Raw_D_metrics/"+Phase+"_HH_raw_Day_metrics_"+str(Houshold)+"_1_exact_3.55555.csv"
Range = np.arange( 20, 60,1)

EXACT_tester = pd.read_csv(Path_CE_MErge,skiprows = 2 )
TEMp = EXACT_tester.iloc[:,4]
Usage = EXACT_tester.iloc[:,3]

print('hafskfhkjeahfisekjfahsiufwa',EXACT_tester.iloc[2,0])

if Phase  == ("2N") or Phase == "3N" or Phase == "3N" or Phase == "4N":
    cooking_threshold = 5
    length_decrease = 40
    start_threshold = 1
    end_threshold = -5
    min_CE_length = 5
    window_slope = 8
else:
    cooking_threshold = 1
    length_decrease = 10
    start_threshold = 1
    end_threshold = -5
    min_CE_length = 8
    window_slope = 12

SUM_cooking_sense = []

for a in Range:
    Usage, Fire_start, Fire_end = Functions_malawi.FireFinder(TEMp, Usage, cooking_threshold, length_decrease, start_threshold, end_threshold, a, min_CE_length, window_slope)
    SUM_cooking_sense.append(sum(Usage))

fig = go.Figure()
fig.add_trace(go.Scatter(x=Range, y=SUM_cooking_sense, mode='lines+markers'))
fig.show()

max_stars_repo_name: GrantRoss-Tenki/Malawi-CQC-CSC-OSU-Work

max_issues_repo_name: GrantRoss-Tenki/Malawi-CQC-CSC-OSU-Work

max_forks_repo_name: GrantRoss-Tenki/Malawi-CQC-CSC-OSU-Work


text:
r"""
This module implements differential operators on cylindrical grids

.. autosummary::
   :nosignatures:

   make_laplace
   make_gradient
   make_divergence
   make_vector_gradient
   make_vector_laplace
   make_tensor_divergence

.. codeauthor:: David Zwicker <david.zwicker@ds.mpg.de>
"""

import numba as nb
import numpy as np

from ... import config
from ...tools.docstrings import fill_in_docstring
from ...tools.numba import jit
from ...tools.typing import OperatorType
from ..cylindrical import CylindricalSymGrid

@CylindricalSymGrid.register_operator("laplace", rank_in=0, rank_out=0)
@fill_in_docstring
def make_laplace(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized laplace operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """
    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    dr_2, dz_2 = 1 / grid.discretization ** 2

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def laplace(arr: np.ndarray, out: np.ndarray) -> None:
        """apply laplace operator to array `arr`"""
        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                arr_z_l, arr_z_h = arr[i, j - 1], arr[i, j + 1]
                arr_r_l, arr_r_h = arr[i - 1, j], arr[i + 1, j]
                out[i, j] = (
                    (arr_r_h - 2 * arr[i, j] + arr_r_l) * dr_2
                    + (arr_r_h - arr_r_l) / (2 * i - 1) * dr_2
                    + (arr_z_l - 2 * arr[i, j] + arr_z_h) * dz_2
                )

    return laplace  # type: ignore

@CylindricalSymGrid.register_operator("gradient", rank_in=0, rank_out=1)
@fill_in_docstring
def make_gradient(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized gradient operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """
    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    scale_r, scale_z = 1 / (2 * grid.discretization)

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def gradient(arr: np.ndarray, out: np.ndarray) -> None:
        """apply gradient operator to array `arr`"""
        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                out[0, i, j] = (arr[i + 1, j] - arr[i - 1, j]) * scale_r
                out[1, i, j] = (arr[i, j + 1] - arr[i, j - 1]) * scale_z
                out[2, i, j] = 0  # no phi dependence by definition

    return gradient  # type: ignore

@CylindricalSymGrid.register_operator("gradient_squared", rank_in=0, rank_out=0)
@fill_in_docstring
def make_gradient_squared(
    grid: CylindricalSymGrid, central: bool = True
) -> OperatorType:
    """make a discretized gradient squared operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created
        central (bool):
            Whether a central difference approximation is used for the gradient
            operator. If this is False, the squared gradient is calculated as
            the mean of the squared values of the forward and backward
            derivatives.

    Returns:
        A function that can be applied to an array of values
    """
    # use processing for large enough arrays
    dim_r, dim_z = grid.shape
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    if central:
        # use central differences
        scale_r, scale_z = 0.25 / grid.discretization ** 2

        @jit(parallel=parallel)
        def gradient_squared(arr: np.ndarray, out: np.ndarray) -> None:
            """apply gradient operator to array `arr`"""
            for i in nb.prange(1, dim_r + 1):  # iterate radial points
                for j in range(1, dim_z + 1):  # iterate axial points
                    term_r = (arr[i + 1, j] - arr[i - 1, j]) ** 2
                    term_z = (arr[i, j + 1] - arr[i, j - 1]) ** 2
                    out[i, j] = term_r * scale_r + term_z * scale_z

    else:
        # use forward and backward differences
        scale_r, scale_z = 0.5 / grid.discretization ** 2

        @jit(parallel=parallel)
        def gradient_squared(arr: np.ndarray, out: np.ndarray) -> None:
            """apply gradient operator to array `arr`"""
            for i in nb.prange(1, dim_r + 1):  # iterate radial points
                for j in range(1, dim_z + 1):  # iterate axial points
                    arr_z_l, arr_c, arr_z_h = arr[i, j - 1], arr[i, j], arr[i, j + 1]
                    term_r = (arr[i + 1, j] - arr_c) ** 2 + (arr_c - arr[i - 1, j]) ** 2
                    term_z = (arr_z_h - arr_c) ** 2 + (arr_c - arr_z_l) ** 2
                    out[i, j] = term_r * scale_r + term_z * scale_z

    return gradient_squared  # type: ignore

@CylindricalSymGrid.register_operator("divergence", rank_in=1, rank_out=0)
@fill_in_docstring
def make_divergence(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized divergence operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """
    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    scale_r, scale_z = 1 / (2 * grid.discretization)
    rs = grid.axes_coords[0]

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def divergence(arr: np.ndarray, out: np.ndarray) -> None:
        """apply divergence operator to array `arr`"""
        arr_r, arr_z = arr[0], arr[1]

        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                out[i, j] = (
                    arr_r[i, j] / rs[i - 1]
                    + (arr_r[i + 1, j] - arr_r[i - 1, j]) * scale_r
                    + (arr_z[i, j + 1] - arr_z[i, j - 1]) * scale_z
                )

    return divergence  # type: ignore

@CylindricalSymGrid.register_operator("vector_gradient", rank_in=1, rank_out=2)
@fill_in_docstring
def make_vector_gradient(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized vector gradient operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """
    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    scale_r, scale_z = 1 / (2 * grid.discretization)
    rs = grid.axes_coords[0]

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def vector_gradient(arr: np.ndarray, out: np.ndarray) -> None:
        """apply gradient operator to array `arr`"""
        # assign aliases
        arr_r, arr_z, arr_φ = arr
        out_rr, out_rz, out_rφ = out[0, 0], out[0, 1], out[0, 2]
        out_zr, out_zz, out_zφ = out[1, 0], out[1, 1], out[1, 2]
        out_φr, out_φz, out_φφ = out[2, 0], out[2, 1], out[2, 2]

        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                out_rr[i, j] = (arr_r[i + 1, j] - arr_r[i - 1, j]) * scale_r
                out_φr[i, j] = (arr_φ[i + 1, j] - arr_φ[i - 1, j]) * scale_r
                out_zr[i, j] = (arr_z[i + 1, j] - arr_z[i - 1, j]) * scale_r

                out_rφ[i, j] = -arr_φ[i, j] / rs[i - 1]
                out_φφ[i, j] = arr_r[i, j] / rs[i - 1]
                out_zφ[i, j] = 0

                out_rz[i, j] = (arr_r[i, j + 1] - arr_r[i, j - 1]) * scale_z
                out_φz[i, j] = (arr_φ[i, j + 1] - arr_φ[i, j - 1]) * scale_z
                out_zz[i, j] = (arr_z[i, j + 1] - arr_z[i, j - 1]) * scale_z

    return vector_gradient  # type: ignore

@CylindricalSymGrid.register_operator("vector_laplace", rank_in=1, rank_out=1)
@fill_in_docstring
def make_vector_laplace(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized vector laplace operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """

    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    rs = grid.axes_coords[0]
    dr, dz = grid.discretization
    s1, s2 = 1 / (2 * dr), 1 / dr ** 2
    scale_z = 1 / (dz ** 2)

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def vector_laplace(arr: np.ndarray, out: np.ndarray) -> None:
        """apply vector laplace operator to array `arr`"""
        # assign aliases
        arr_r, arr_z, arr_φ = arr
        out_r, out_z, out_φ = out

        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                f_r_l, f_r_m, f_r_h = arr_r[i - 1, j], arr_r[i, j], arr_r[i + 1, j]
                out_r[i, j] = (
                    (arr_r[i, j + 1] - 2 * f_r_m + arr_r[i, j - 1]) * scale_z
                    - f_r_m / rs[i - 1] ** 2
                    + (f_r_h - f_r_l) * s1 / rs[i - 1]
                    + (f_r_h - 2 * f_r_m + f_r_l) * s2
                )

                f_φ_l, f_φ_m, f_φ_h = arr_φ[i - 1, j], arr_φ[i, j], arr_φ[i + 1, j]
                out_φ[i, j] = (
                    (arr_φ[i, j + 1] - 2 * f_φ_m + arr_φ[i, j - 1]) * scale_z
                    - f_φ_m / rs[i - 1] ** 2
                    + (f_φ_h - f_φ_l) * s1 / rs[i - 1]
                    + (f_φ_h - 2 * f_φ_m + f_φ_l) * s2
                )

                f_z_l, f_z_m, f_z_h = arr_z[i - 1, j], arr_z[i, j], arr_z[i + 1, j]
                out_z[i, j] = (
                    (arr_z[i, j + 1] - 2 * f_z_m + arr_z[i, j - 1]) * scale_z
                    + (f_z_h - f_z_l) * s1 / rs[i - 1]
                    + (f_z_h - 2 * f_z_m + f_z_l) * s2
                )

    return vector_laplace  # type: ignore

@CylindricalSymGrid.register_operator("tensor_divergence", rank_in=2, rank_out=1)
@fill_in_docstring
def make_tensor_divergence(grid: CylindricalSymGrid) -> OperatorType:
    """make a discretized tensor divergence operator for a cylindrical grid

    {DESCR_CYLINDRICAL_GRID}

    Args:
        grid (:class:`~pde.grids.cylindrical.CylindricalSymGrid`):
            The grid for which the operator is created

    Returns:
        A function that can be applied to an array of values
    """
    # calculate preliminary quantities
    dim_r, dim_z = grid.shape
    rs = grid.axes_coords[0]
    scale_r, scale_z = 1 / (2 * grid.discretization)

    # use processing for large enough arrays
    parallel = dim_r * dim_z >= config["numba.parallel_threshold"]

    @jit(parallel=parallel)
    def tensor_divergence(arr: np.ndarray, out: np.ndarray) -> None:
        """apply tensor divergence operator to array `arr`"""
        # assign aliases
        arr_rr, arr_rz, arr_rφ = arr[0, 0], arr[0, 1], arr[0, 2]
        arr_zr, arr_zz, _ = arr[1, 0], arr[1, 1], arr[1, 2]
        arr_φr, arr_φz, arr_φφ = arr[2, 0], arr[2, 1], arr[2, 2]
        out_r, out_z, out_φ = out

        for i in nb.prange(1, dim_r + 1):  # iterate radial points
            for j in range(1, dim_z + 1):  # iterate axial points
                out_r[i, j] = (
                    (arr_rz[i, j + 1] - arr_rz[i, j - 1]) * scale_z
                    + (arr_rr[i + 1, j] - arr_rr[i - 1, j]) * scale_r
                    + (arr_rr[i, j] - arr_φφ[i, j]) / rs[i - 1]
                )

                out_φ[i, j] = (
                    (arr_φz[i, j + 1] - arr_φz[i, j - 1]) * scale_z
                    + (arr_φr[i + 1, j] - arr_φr[i - 1, j]) * scale_r
                    + (arr_rφ[i, j] + arr_φr[i, j]) / rs[i - 1]
                )

                out_z[i, j] = (
                    (arr_zz[i, j + 1] - arr_zz[i, j - 1]) * scale_z
                    + (arr_zr[i + 1, j] - arr_zr[i - 1, j]) * scale_r
                    + arr_zr[i, j] / rs[i - 1]
                )

    return tensor_divergence  # type: ignore

max_stars_repo_name: mfkiwl/py-pde

max_issues_repo_name: mfkiwl/py-pde

max_forks_repo_name: mfkiwl/py-pde

reason: import numpy,import numba


text:
"""Sparse 0.11.0 broke nmrsim. This test suite contains smoke tests that
third-party modules (sparse/numba/numpy) are still behaving as intended.

When this test suite was created, the scipy version of tensordot was failing,
even for sparse 0.10.0.

Because sparse is not Version 1 yet, tests for its functions are a good idea.
"""
import numpy as np
import pytest
import sparse
from sparse import COO
from sparse._utils import assert_eq  # noqa

from .qm_arguments import spin2

# The following three tests are copied over from the sparse project.
# They pass in sparse >=0.11.0
@pytest.mark.skipif(sparse.__version__ < "0.11.0",
                    reason="This test from the sparse library requires v0.11 or greater")
@pytest.mark.parametrize(
    "a_shape,b_shape,axes",
    [
        [(3, 4), (4, 3), (1, 0)],
        [(3, 4), (4, 3), (0, 1)],
        [(3, 4, 5), (4, 3), (1, 0)],
        [(3, 4), (5, 4, 3), (1, 1)],
        [(3, 4), (5, 4, 3), ((0, 1), (2, 1))],
        [(3, 4), (5, 4, 3), ((1, 0), (1, 2))],
        [(3, 4, 5), (4,), (1, 0)],
        [(4,), (3, 4, 5), (0, 1)],
        [(4,), (4,), (0, 0)],
        [(4,), (4,), 0],
    ],
)
@pytest.mark.parametrize(
    "a_format, b_format",
    [("coo", "coo"), ("coo", "gcxs"), ("gcxs", "coo"), ("gcxs", "gcxs")],
)
def test_tensordot(a_shape, b_shape, axes, a_format, b_format):
    from sparse._compressed import GCXS
    # else:
    #     assert 1 == 2
    sa = sparse.random(a_shape, density=0.5, format=a_format)
    sb = sparse.random(b_shape, density=0.5, format=b_format)

    a = sa.todense()
    b = sb.todense()

    a_b = np.tensordot(a, b, axes)

    # tests for return_type=None
    sa_sb = sparse.tensordot(sa, sb, axes)
    sa_b = sparse.tensordot(sa, b, axes)
    a_sb = sparse.tensordot(a, sb, axes)

    assert_eq(a_b, sa_sb)
    assert_eq(a_b, sa_b)
    assert_eq(a_b, a_sb)
    if all(isinstance(arr, COO) for arr in [sa, sb]):
        assert isinstance(sa_sb, COO)
    else:
        assert isinstance(sa_sb, GCXS)
    assert isinstance(sa_b, np.ndarray)
    assert isinstance(a_sb, np.ndarray)

    # tests for return_type=COO
    sa_b = sparse.tensordot(sa, b, axes, return_type=COO)
    a_sb = sparse.tensordot(a, sb, axes, return_type=COO)

    assert_eq(a_b, sa_b)
    assert_eq(a_b, a_sb)
    assert isinstance(sa_b, COO)
    assert isinstance(a_sb, COO)

    # tests form return_type=GCXS
    sa_b = sparse.tensordot(sa, b, axes, return_type=GCXS)
    a_sb = sparse.tensordot(a, sb, axes, return_type=GCXS)

    assert_eq(a_b, sa_b)
    assert_eq(a_b, a_sb)
    assert isinstance(sa_b, GCXS)
    assert isinstance(a_sb, GCXS)

    # tests for return_type=np.ndarray
    sa_sb = sparse.tensordot(sa, sb, axes, return_type=np.ndarray)

    assert_eq(a_b, sa_sb)
    assert isinstance(sa_sb, np.ndarray)

def test_tensordot_empty():
    x1 = np.empty((0, 0, 0))
    x2 = np.empty((0, 0, 0))
    s1 = sparse.COO.from_numpy(x1)
    s2 = sparse.COO.from_numpy(x2)

    assert_eq(np.tensordot(x1, x2), sparse.tensordot(s1, s2))

def test_tensordot_valueerror():
    x1 = sparse.COO(np.array(1))
    x2 = sparse.COO(np.array(1))

    with pytest.raises(ValueError):
        x1 @ x2

def test_numpy_tensordot():
    """A stepping stone to finding the problem with sparse 0.11.0's tensordot.

    This is taken from the Numpy tensordot documentation ("traditional" example):
    https://numpy.org/doc/stable/reference/generated/numpy.tensordot.html
    """
    # GIVEN accepted numpy doc example for tensordot
    a = np.arange(60.).reshape(3, 4, 5)  # noqa
    b = np.arange(24.).reshape(4, 3, 2)  # noqa
    # WHEN the following tensordot is performed
    c = np.tensordot(a, b, axes=([1, 0], [0, 1]))
    # THEN the expected results are seen
    assert c.shape == (5, 2)
    assert np.array_equal(c,
                          np.array(
                              [[4400., 4730.],
                               [4532., 4874.],
                               [4664., 5018.],
                               [4796., 5162.],
                               [4928., 5306.]]
                          ))

@pytest.mark.xfail(reason="sparse bug")
def test_sparse_tensordot():
    """Tests to see if sparse_tensordot can pass the numpy tensordot test above.
    Fails for all sparse versions. See:
    https://github.com/pydata/sparse/issues/493

    nmrsim luckily always worked prior to sparse v0.11 because
    np arrays were never used as inputs for sparse.tensordot,
    so this bug with sparse was never encountered.
    """
    # GIVEN accepted numpy doc example for tensordot
    a = np.arange(60.).reshape(3, 4, 5)  # noqa
    b = np.arange(24.).reshape(4, 3, 2)  # noqa
    # WHEN the sparse version of tensordot is performed
    c_sparse = sparse.tensordot(a, b, axes=([1, 0], [0, 1]))
    # THEN the expected results are seen
    c = c_sparse.todense()
    assert c.shape == (5, 2)
    assert np.array_equal(c,
                          np.array(
                              [[4400., 4730.],
                               [4532., 4874.],
                               [4664., 5018.],
                               [4796., 5162.],
                               [4928., 5306.]]
                          ))

@pytest.mark.xfail(reason="sparse bug will cause second parametrized test to fail")
@pytest.mark.parametrize("tensordot", [np.tensordot, sparse.tensordot])
def test_twospin_v(tensordot):
    """A test using one of the sub-calculations for the Hamiltonian
    to see if sparse.tensordot can do the same calculation as np.tensordot.

    Redundant with the above two tests.
    """
    v, J = spin2()
    Lz = np.array(
        [[[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, 0.5 + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, 0. + 0.j, -0.5 + 0.j, -0. + 0.j],
          [0. + 0.j, 0. + 0.j, -0. + 0.j, -0.5 + 0.j]],

         [[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, -0.5 + 0.j, 0. + 0.j, -0. + 0.j],
          [0. + 0.j, 0. + 0.j, 0.5 + 0.j, 0. + 0.j],
          [0. + 0.j, -0. + 0.j, 0. + 0.j, -0.5 + 0.j]]]
    )

    H = tensordot(v, Lz, axes=1)
    assert np.allclose(H,
                       np.array(
                           [[15. + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
                            [0. + 0.j, -5. + 0.j, 0. + 0.j, 0. + 0.j],
                            [0. + 0.j, 0. + 0.j, 5. + 0.j, 0. + 0.j],
                            [0. + 0.j, 0. + 0.j, 0. + 0.j, -15. + 0.j]]
                       ))

def test_twospin_v_coo():
    """Tests to see if sparse.tensordot works with COO arrays instead.

    This test passes for sparse <=0.10.0, but fails for >=0.11.0,
    and generates the same nmrsim error that was observed when sparse was upgraded.
    """
    v, J = spin2()
    Lz = np.array(
        [[[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, 0.5 + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, 0. + 0.j, -0.5 + 0.j, -0. + 0.j],
          [0. + 0.j, 0. + 0.j, -0. + 0.j, -0.5 + 0.j]],

         [[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
          [0. + 0.j, -0.5 + 0.j, 0. + 0.j, -0. + 0.j],
          [0. + 0.j, 0. + 0.j, 0.5 + 0.j, 0. + 0.j],
          [0. + 0.j, -0. + 0.j, 0. + 0.j, -0.5 + 0.j]]]
    )
    v_coo = sparse.COO(v)
    Lz_coo = sparse.COO(Lz)
    H = sparse.tensordot(v_coo, Lz_coo, axes=1)
    assert np.allclose(
        H.todense(),
        np.array(
            [[15. + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
             [0. + 0.j, -5. + 0.j, 0. + 0.j, 0. + 0.j],
             [0. + 0.j, 0. + 0.j, 5. + 0.j, 0. + 0.j],
             [0. + 0.j, 0. + 0.j, 0. + 0.j, -15. + 0.j]]
        ))

# v2, J2 = spin2()

v2 = np.array([10.0, 20.0])
J2 = None

Lz2 = np.array(
    [[[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
      [0. + 0.j, 0.5 + 0.j, 0. + 0.j, 0. + 0.j],
      [0. + 0.j, 0. + 0.j, -0.5 + 0.j, -0. + 0.j],
      [0. + 0.j, 0. + 0.j, -0. + 0.j, -0.5 + 0.j]],

     [[0.5 + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
      [0. + 0.j, -0.5 + 0.j, 0. + 0.j, -0. + 0.j],
      [0. + 0.j, 0. + 0.j, 0.5 + 0.j, 0. + 0.j],
      [0. + 0.j, -0. + 0.j, 0. + 0.j, -0.5 + 0.j]]]
)
v2_coo = sparse.COO(v2)
Lz2_coo = sparse.COO(Lz2)

@pytest.mark.xfail(reason="sparse bug will cause double-ndarray case to fail")
@pytest.mark.parametrize("v, Lz", [
    (v2, Lz2),
    (v2_coo, Lz2),
    (v2, Lz2_coo),
    (v2_coo, Lz2_coo)
])
def test_twospin_v_Lz(v, Lz):
    """Tests to see what combinations of argument types cause which errors.

    Different errors for: both ndarray; one ndarray and one COO; both COO.
    See: https://github.com/pydata/sparse/issues/499
    """
    H = sparse.tensordot(v, Lz, axes=1)
    if not isinstance(H, np.ndarray):
        H = H.todense()
    assert np.allclose(H,
                       np.array(
                           [[15. + 0.j, 0. + 0.j, 0. + 0.j, 0. + 0.j],
                            [0. + 0.j, -5. + 0.j, 0. + 0.j, 0. + 0.j],
                            [0. + 0.j, 0. + 0.j, 5. + 0.j, 0. + 0.j],
                            [0. + 0.j, 0. + 0.j, 0. + 0.j, -15. + 0.j]]
                       ))

a = np.arange(60.).reshape(3, 4, 5)  # noqa
b = np.arange(24.).reshape(4, 3, 2)  # noqa
a_coo = sparse.COO(a)
b_coo = sparse.COO(b)

@pytest.mark.xfail(reason="sparse bug will cause double ndarray case to fail")
@pytest.mark.parametrize("a, b", [
    (a, b),
    (a_coo, b),
    (a, b_coo),
    (a_coo, b_coo)
])
def test_sparse_tensordot_inputs(a, b):
    """Tests the numpy "typical" example versus different argument type
    combinations.

    As in other tests, double-ndarray gives an error across all versions of
    sparse; one ndarray and one COO gives one numba error; and both COO gives
    another numba error. See: https://github.com/pydata/sparse/issues/499
    """
    # GIVEN accepted numpy doc example for tensordot

    # WHEN the sparse version of tensordot is performed
    c = sparse.tensordot(a, b, axes=([1, 0], [0, 1]))

    # THEN the expected results are seen
    if not isinstance(c, np.ndarray):
        c = c.todense()
    assert c.shape == (5, 2)
    assert np.array_equal(c,
                          np.array(
                              [[4400., 4730.],
                               [4532., 4874.],
                               [4664., 5018.],
                               [4796., 5162.],
                               [4928., 5306.]]
                          ))

max_stars_repo_name: sametz/nmrsim

max_issues_repo_name: sametz/nmrsim

max_forks_repo_name: sametz/nmrsim


text:
from __future__ import division

import numpy as np
from resippy.spectral.spectrum import Spectrum

WAVELENGTH_UNITS = "nanometers"
SPECTRUM_UNITS = "reflectance_0-1"

class SvcAsciiFileFactory:
    @classmethod
    def from_sig_file(cls, ascii_file     # type: str
                          ):                  # type: (...) -> Spectrum
        with open(ascii_file) as f:
            content = f.readlines()

        data_ind = 0
        for ind in range(len(content)):
            data_ind += 1
            if content[ind].startswith('data'):
                break

        wavelengths = []
        spectral_data = []
        while data_ind < len(content):
            wave, _, _, ref = content[data_ind].split()

            data_ind += 1
            wavelengths.append(float(wave))
            spectral_data.append(float(ref)/100.0)

        spectral_data = np.array(spectral_data)
        wavelengths = np.array(wavelengths)

        spectrum = Spectrum()
        spectrum.set_wavelength_units(WAVELENGTH_UNITS)
        spectrum.set_wavelengths(wavelengths)
        spectrum.set_spectral_data(spectral_data)
        spectrum.set_spectrum_units(SPECTRUM_UNITS)
        return spectrum

max_stars_repo_name: BeamIO-Inc/resippy

max_issues_repo_name: BeamIO-Inc/resippy

max_forks_repo_name: BeamIO-Inc/resippy


text:
#!/usr/bin/env python3

import os
import sys
from shutil import copyfile

import numpy as np

from skimage import io

import hasel

import pystache

from config import get_args
from clustering import hhsl_cluster_centers_as_hsl, hsl_cluster_centers
from converters import hex2rgb, rgb2hex, rgblist2hex, hsllist2hex, hsl2rgb, hsl2hex
from htmlpreview import get_html_contents
from scoring import pick_n_best_colors, clip_between_boundaries, find_dominant_by_frequency, sort_colors_by_closest_counterpart, pick_n_best_colors_with_reference
from colorgenerator import generate_complementary, generate_similar, correct_saturation
from logger import log

n_colors = 16 # must be less than or equal to n_clusters

def get_pixels_for_image(img_file_path):
    log("reading image \"" + img_file_path + "\"")
    img_rgb = io.imread(img_file_path)

    log("converting color space...")

    if (len(img_rgb.shape) < 3): # this is a greyscale image
        w, h = img_rgb.shape
        ret = np.empty((w, h, 3), dtype=np.uint8)
        ret[:, :, 0] = img_rgb
        ret[:, :, 1] = img_rgb
        ret[:, :, 2] = img_rgb
        img_rgb = ret

    img_hsl = hasel.rgb2hsl(img_rgb[:,:,0:3])
    hsl_colors = img_hsl.reshape((-1, 3))

    return hsl_colors

dir_path = os.path.expanduser("~/.gvcci")
try:
    os.stat(dir_path)
except:
    os.mkdir(dir_path)

dir_path = os.path.expanduser("~/.gvcci/themes")
try:
    os.stat(dir_path)
except:
    os.mkdir(dir_path)

html_contents = ""

args = get_args()

images = args['images']
background = args['background']
template_path = args['template_path']
print_output = args['print_output']
symlink_wallpaper = args['symlink_wallpaper']

with open('resources/gvcci-title-ascii.txt', 'r') as logo:
    log(logo.read())

image_paths = map(os.path.realpath, images)

for img_file_path in image_paths:
    log("Generating colors for input " + str(img_file_path))

    hsl_colors = get_pixels_for_image(img_file_path)
    improved_centers = hhsl_cluster_centers_as_hsl(hsl_colors)

    dominant_dark_and_light_colors = find_dominant_by_frequency(hsl_colors)

    bg_color = dominant_dark_and_light_colors[0]
    fg_color = dominant_dark_and_light_colors[1]

    dominant_dark = dominant_dark_and_light_colors[0]
    dominant_light = dominant_dark_and_light_colors[1]

    if (dominant_dark[0][2] > dominant_light[0][2]):
        tmp = dominant_light
        dominant_light = dominant_dark
        dominant_dark = tmp

    max_dominant_dark_saturation = 0.4

    reference_dominant_light_color = np.array([[0, 0, 0.94]])
    reference_dominant_light_color = np.array([[0, 0, min(max(dominant_light[0][2], 0.8), 0.95)]])

    if background == "dark" or (background == "auto" and bg_color[0][2] < 0.5):
        if (dominant_dark[0][1] > max_dominant_dark_saturation):
            dominant_dark[0][1] = max_dominant_dark_saturation
        bg_color = dominant_dark
        fg_color = dominant_light
    elif background == "light" or (background == "auto" and bg_color[0][2] >= 0.5):
        dominant_light = correct_saturation(dominant_light)
        bg_color = dominant_light
        fg_color = dominant_dark
    elif background[0] == "#":
        bg_color = hex2rgb(background)
        bg_color = hasel.rgb2hsl(np.array(bg_color).reshape(1, 1, 3)).reshape(1, 3)
        if (bg_color[0][2] < 0.5):
            fg_color = dominant_light
        else:
            fg_color = dominant_dark

    # Accessibility contrast levels:
    # WCAG 2.0 level AA requires a contrast ratio of [4.5 : 1] for normal text
    # WCAG 2.0 level AAA requires a contrast ratio of [7 : 1] for normal text
    # Note: the contrast maxes out at 21.0 for white / black contrast

    # dark theme settings
    min_dark_contrast = 5
    min_light_contrast = 1.7

    # light theme settings
    if (bg_color[0][2] > 0.5):
        min_dark_contrast = 16
        min_light_contrast = 3

    # ansi constants
    black   = [0,       0, 0. ]
    red     = [0,       1, 0.5]
    green   = [0.33333, 1, 0.4]
    yellow  = [0.16666, 1, 0.5]
    blue    = [0.66666, 1, 0.6]
    magenta = [0.83333, 1, 0.5]
    cyan    = [0.5,     1, 0.5]
    white   = [0,       0, 1.0]

    standard_ansi_colors = np.array([
        red,
        green,
        yellow,
        blue,
        magenta,
        cyan,
        white,
        black
    ])

    # ansi_colors_unconstrained = pick_n_best_colors_with_reference(8, improved_centers, standard_ansi_colors, dominant_dark, dominant_light, min_dark_contrast, min_light_contrast)
    ansi_colors_unconstrained = pick_n_best_colors(8, improved_centers, dominant_dark, dominant_light, min_dark_contrast, min_light_contrast)
    ansi_colors_normal = clip_between_boundaries(ansi_colors_unconstrained, dominant_dark, dominant_light, min_dark_contrast, min_light_contrast)
    ansi_colors_sorted = sort_colors_by_closest_counterpart(ansi_colors_normal, standard_ansi_colors)
    # ansi_colors_sorted = ansi_colors_normal # already sorted when you run it through pick_n_best_with_ref
    ansi_colors_normal_and_bright = generate_complementary(ansi_colors_sorted)
    ansi_colors = ansi_colors_normal_and_bright

    black = bg_color.copy()

    if (bg_color[0][2] < 0.1):
        black[0][2] += 0.1
    elif (bg_color[0][2] < 0.5):
        black[0][2] -= 0.1
    else:
        black[0][2] = 0.2

    black_bright = black.copy()
    black_bright[0][2] += 0.1

    white_l_threshold = 0.85
    if ansi_colors[12][2] < white_l_threshold:
        ansi_colors[12][2] = white_l_threshold # white l value lower bound
        ansi_colors[13][2] = white_l_threshold + 0.12

    html_contents += get_html_contents(np.vstack((black, black_bright, ansi_colors)), bg_color, fg_color, img_file_path)
    html =  "\n"
    html += "
"
    html += html_contents
    html += "
"
    html += "\n"

    result_file = open("examples.html", "w")
    result_file.write(html)
    result_file.close()

    colors_hsl = {
        "background":          bg_color,
        "foreground":          fg_color,
        "bold":                fg_color, # TODO!
        "cursor":              ansi_colors[14],
        "selection":           ansi_colors[15],
        "selected-text":       bg_color,
        "ansi-black-normal":   black,
        "ansi-black-bright":   black_bright,
        "ansi-red-normal":     ansi_colors[0],
        "ansi-red-bright":     ansi_colors[1],
        "ansi-green-normal":   ansi_colors[2],
        "ansi-green-bright":   ansi_colors[3],
        "ansi-yellow-normal":  ansi_colors[4],
        "ansi-yellow-bright":  ansi_colors[5],
        "ansi-blue-normal":    ansi_colors[6],
        "ansi-blue-bright":    ansi_colors[7],
        "ansi-magenta-normal": ansi_colors[8],
        "ansi-magenta-bright": ansi_colors[9],
        "ansi-cyan-normal":    ansi_colors[10],
        "ansi-cyan-bright":    ansi_colors[11],
        "ansi-white-normal":   ansi_colors[12],
        "ansi-white-bright":   ansi_colors[13]
    }

    colors = {}
    for name, hsl in colors_hsl.items():
        rgb = hsl2rgb(hsl)
        hex = hsl2hex(hsl)

        colors[name + "-red-255"]     = rgb[0]
        colors[name + "-green-255"]   = rgb[1]
        colors[name + "-blue-255"]    = rgb[2]
        colors[name + "-red-float"]   = rgb[0] / 255
        colors[name + "-green-float"] = rgb[1] / 255
        colors[name + "-blue-float"]  = rgb[2] / 255
        colors[name + "-hex"]         = hex

    log("=========== Terminal Colors ===========")
    with open('templates/columns-with-headers.txt', 'r') as print_template:
        log(pystache.render(print_template.read(), colors))
    log("=======================================")

    image_name = os.path.basename(img_file_path).split(".")[0]
    image_extension = os.path.basename(img_file_path).split(".")[-1]

    output_dir_path = os.path.join(os.path.expanduser("~/.gvcci/themes"), image_name)
    try:
        os.stat(output_dir_path)
    except:
        os.mkdir(output_dir_path)

    output_image_path = os.path.join(output_dir_path, "wallpaper")

    try:
        os.remove(output_image_path)
    except OSError:
        pass

    if (symlink_wallpaper):
        os.symlink(img_file_path, output_image_path)
    else:
        copyfile(img_file_path, output_image_path)

    log("Output: " + output_image_path)

    template_file_or_dir_path = os.path.realpath(template_path)
    if (os.path.isfile(template_file_or_dir_path)):
        template_file_path = template_file_or_dir_path
        with open(template_file_path, 'r') as template_file:
            template = template_file.read()
            template_file_name = os.path.basename(template_file_path)

            output_theme_path = os.path.join(output_dir_path, template_file_name)

            with open(output_theme_path, 'w') as out_file:
                filled_out_template = pystache.render(template, colors)
                out_file.write(filled_out_template)
                if (print_output):
                    print(filled_out_template)

            log("Output: " + output_theme_path)
    elif (os.path.isdir(template_file_or_dir_path)):
        for template_file in os.listdir(template_file_or_dir_path):
            template_file_path = os.path.join(template_file_or_dir_path, template_file)
            with open(template_file_path, 'r') as template_file:
                template = template_file.read()
                template_file_name = os.path.basename(template_file_path)

                output_theme_path = os.path.join(output_dir_path, template_file_name)

                with open(output_theme_path, 'w') as out_file:
                    filled_out_template = pystache.render(template, colors)
                    out_file.write(filled_out_template)
                    if (print_output):
                        print(filled_out_template)

                log("Output: " + output_theme_path)

    log("=======================================")

max_stars_repo_name: rschmukler/gvcci

max_issues_repo_name: rschmukler/gvcci

max_forks_repo_name: rschmukler/gvcci


text:
import copy
from typing import Any, Dict, List, Optional, Union

import numpy as np
import pandas as pd
from scipy.stats import chisquare

def _chi2(bad_rates: List[Dict], overall_rate: float) -> float:
    f_obs = [_bin["bad"] for _bin in bad_rates]
    f_exp = [_bin["total"] * overall_rate for _bin in bad_rates]

    chi2 = chisquare(f_obs=f_obs, f_exp=f_exp)[0]

    return chi2

def _check_diff_woe(bad_rates: List[Dict], diff_woe_threshold: float) -> Union[None, int]:
    woe_delta: np.ndarray = np.abs(np.diff([bad_rate["woe"] for bad_rate in bad_rates]))
    min_diff_woe = min(sorted(list(set(woe_delta))))
    if min_diff_woe < diff_woe_threshold:
        return list(woe_delta).index(min_diff_woe)
    else:
        return None

def _mono_flags(bad_rates: List[Dict]) -> bool:
    bad_rate_diffs = np.diff([bad_rate["bad_rate"] for bad_rate in bad_rates])
    positive_mono_diff = np.all(bad_rate_diffs > 0)
    negative_mono_diff = np.all(bad_rate_diffs < 0)
    return True in [positive_mono_diff, negative_mono_diff]

def _find_index_of_diff_flag(bad_rates: List[Dict]) -> int:
    bad_rate_diffs = np.diff([bad_rate["bad_rate"] for bad_rate in bad_rates])
    idx = list(bad_rate_diffs > 0).index(pd.Series(bad_rate_diffs > 0).value_counts().sort_values().index.tolist()[0])
    return idx

def _merge_bins_chi(x: np.ndarray, y: np.ndarray, bad_rates: List[Dict], bins: List):
    idx = _find_index_of_diff_flag(bad_rates)
    if idx == 0:
        del bins[1]
    elif idx == len(bad_rates) - 2:
        del bins[len(bins) - 2]
    else:
        temp_bins = copy.deepcopy(bins)
        del temp_bins[idx + 1]
        temp_bad_rates, temp_overall_rate = bin_bad_rate(x, y, temp_bins)
        chi_1 = _chi2(temp_bad_rates, temp_overall_rate)
        del temp_bins

        temp_bins = copy.deepcopy(bins)
        del temp_bins[idx + 2]
        temp_bad_rates, temp_overall_rate = bin_bad_rate(x, y, temp_bins)
        chi_2 = _chi2(temp_bad_rates, temp_overall_rate)
        if chi_1 < chi_2:
            del bins[idx + 1]
        else:
            del bins[idx + 2]
    bad_rates, _ = bin_bad_rate(x, y, bins)
    return bad_rates, bins

def _merge_bins_min_pct(
        x: np.ndarray, y: np.ndarray, bad_rates: List[Dict], bins: List, cat: bool = False
):
    idx = [
        pct for pct in [bad_rates[i]["pct"] for i in range(len(bad_rates))]
    ].index(min([bad_rate["pct"] for bad_rate in bad_rates]))

    if cat:
        if idx == 0:
            bins[idx + 1] += bins[idx]
        elif idx == len(bad_rates) - 1:
            bins[idx - 1] += bins[idx]
        else:
            if bad_rates[idx - 1]["pct"] < bad_rates[idx + 1]["pct"]:
                bins[idx - 1] += bins[idx]
            else:
                bins[idx + 1] += bins[idx]
        del bins[idx]
    else:
        if idx == 0:
            del bins[1]
        elif idx == len(bad_rates) - 1:
            del bins[len(bins) - 2]
        else:
            if bad_rates[idx - 1]["pct"] < bad_rates[idx + 1]["pct"]:
                del bins[idx]
            else:
                del bins[idx + 1]

    bad_rates, _ = bin_bad_rate(x, y, bins, cat=cat)
    if cat:
        bins = [bad_rate["bin"] for bad_rate in bad_rates]
    return bad_rates, bins

def bin_bad_rate(
        x: np.ndarray, y: np.ndarray, bins: List, cat: bool = False
) -> tuple[list[dict[str, Union[Union[list, int, float], Any]]], Union[Optional[float], Any]]:
    bad_rates = []
    if cat:
        max_idx = len(bins)
    else:
        max_idx = len(bins) - 1
    for i in range(max_idx):
        if cat:
            value = bins[i]
        else:
            value = [bins[i], bins[i + 1]]

        x_not_na = x[~pd.isna(x)]
        y_not_na = y[~pd.isna(x)]
        if cat:
            x_in = x_not_na[pd.Series(x_not_na).isin(value)]
        else:
            x_in = x_not_na[
                np.where((x_not_na >= np.min(value)) & (x_not_na < np.max(value)))
            ]
        total = len(x_in)
        all_bad = y[~pd.isna(x)].sum()
        all_good = len(y[~pd.isna(x)]) - all_bad
        bad = y_not_na[np.isin(x_not_na, x_in)].sum()
        pct = np.sum(np.isin(x_not_na, x_in)) * 1.0 / len(x)
        bad_rate = bad / total
        good = total - bad
        good_rate = good / total
        if good != 0 and bad != 0:
            woe = np.log((good / all_good) / (bad / all_bad))
        else:
            woe = np.log((good + 0.5 / all_good) / (bad + 0.5 / all_bad))
        iv = (good_rate - bad_rate) * woe
        stats = {
            "bin": value,
            "total": total,
            "bad": bad,
            "pct": pct,
            "bad_rate": bad_rate,
            "woe": woe,
            "iv": iv,
        }
        bad_rates.append(stats)

    if cat:
        bad_rates.sort(key=lambda _x: _x["bad_rate"])

    overall_rate = None
    if not cat:
        bad = sum([bad_rate["bad"] for bad_rate in bad_rates])
        total = sum([bad_rate["total"] for bad_rate in bad_rates])

        overall_rate = bad * 1.0 / total

    return bad_rates, overall_rate

def cat_binning(
        x: np.ndarray,
        y: np.ndarray,
        min_pct_group: float,
        max_bins: int,
        diff_woe_threshold: float,
) -> tuple[list[dict[str, Union[Union[list, int, float], Any]]], Optional[str]]:
    missing_bin = None

    try:
        x = x.astype(float)
        data_type = "float"
    except ValueError:
        x = x.astype(object)
        data_type = "object"

    bins = list([bin] for bin in np.unique(x[~pd.isna(x)]))

    if len(bins) > max_bins:
        bad_rates_dict = dict(
            sorted(
                {bins[i][0]: y[np.isin(x, bins[i])].sum() / len(y[np.isin(x, bins[i])]) for i in
                 range(len(bins))}.items(), key=lambda item: item[1]
            )
        )
        bad_rate_list = [bad_rates_dict[i] for i in bad_rates_dict]
        q_list = [0.0]
        for quantile in range(1, max_bins):
            q_list.append(
                np.nanquantile(np.array(bad_rate_list), quantile / max_bins, axis=0)
            )
        q_list.append(1)
        q_list = list(sorted(set(q_list)))

        new_bins = [copy.deepcopy([list(bad_rates_dict.keys())[0]])]
        start = 1
        for i in range(len(q_list) - 1):
            for n in range(start, len(list(bad_rates_dict.keys()))):
                if bad_rate_list[n] >= q_list[i + 1]:
                    break
                elif (bad_rate_list[n] >= q_list[i]) & (
                        bad_rate_list[n] < q_list[i + 1]
                ):
                    try:
                        new_bins[i] += [list(bad_rates_dict.keys())[n]]
                        start += 1
                    except IndexError:
                        new_bins.append([])
                        new_bins[i] += [list(bad_rates_dict.keys())[n]]
                        start += 1

        bad_rates, _ = bin_bad_rate(x, y, new_bins, cat=True)
        bins = [bad_rate["bin"] for bad_rate in bad_rates]
    else:
        bad_rates, _ = bin_bad_rate(x, y, bins, cat=True)

    if len(y[pd.isna(x)]) > 0:
        if len(bins) < 2:
            bins.append([])
            if data_type == "object":
                bins[1] += ["Missing"]
                x[pd.isna(x)] = "Missing"
            else:
                bins[1] += [-1]
                x[pd.isna(x)] = -1
            bad_rates, _ = bin_bad_rate(x, y, bins, cat=True)
            if bad_rates[0]["bin"][0] == "Missing" or bad_rates[0]["bin"][0] == -1:
                missing_bin = "first"
            else:
                missing_bin = "last"
        else:
            na_bad_rate = y[pd.isna(x)].sum() / len(y[pd.isna(x)])
            if abs(na_bad_rate - bad_rates[0]["bad_rate"]) < abs(
                    na_bad_rate - bad_rates[len(bad_rates) - 1]["bad_rate"]
            ):
                missing_bin = "first"
                if data_type == "object":
                    bad_rates[0]["bin"] += ["Missing"]
                    x[pd.isna(x)] = "Missing"
                else:
                    bad_rates[0]["bin"] += [-1]
                    x[pd.isna(x)] = -1
            else:
                missing_bin = "last"
                if data_type == "object":
                    bad_rates[-1]["bin"] += ["Missing"]
                    x[pd.isna(x)] = "Missing"
                else:
                    bad_rates[-1]["bin"] += [-1]
                    x[pd.isna(x)] = -1
            bad_rates, _ = bin_bad_rate(x, y, bins, cat=True)
            bins = [bad_rate["bin"] for bad_rate in bad_rates]

    if len(bins) <= 2:
        return bad_rates, missing_bin

    while (_check_diff_woe(bad_rates, diff_woe_threshold) is not None) and (
            len(bad_rates) > 2
    ):
        idx = _check_diff_woe(bad_rates, diff_woe_threshold)
        bins[idx + 1] += bins[idx]
        del bins[idx]
        bad_rates, _ = bin_bad_rate(x, y, bins, cat=True)
        bins = [bad_rate["bin"] for bad_rate in bad_rates]
        idx = _check_diff_woe(bad_rates, diff_woe_threshold)

    if len(bins) <= 2:
        return bad_rates, missing_bin

    while (
            min([bad_rate["pct"] for bad_rate in bad_rates]) <= min_pct_group
            and len(bins) > 2
    ):
        bad_rates, bins = _merge_bins_min_pct(x, y, bad_rates, bins, cat=True)
        bins = [bad_rate["bin"] for bad_rate in bad_rates]

    while len(bad_rates) > max_bins and len(bins) > 2:
        bad_rates, bins = _merge_bins_min_pct(x, y, bad_rates, bins, cat=True)
        bins = [bad_rate["bin"] for bad_rate in bad_rates]

    return bad_rates, missing_bin

def num_binning(
        x: np.ndarray,
        y: np.ndarray,
        min_pct_group: float,
        max_bins: int,
        diff_woe_threshold: float,
) -> tuple[list[dict[str, Union[Union[list, int, float], Any]]], Optional[str]]:
    missing_bin = None
    bins = [np.NINF]
    if len(np.unique(x[~pd.isna(x)])) > max_bins:
        for quantile in range(1, max_bins):
            bins.append(np.nanquantile(x, quantile / max_bins, axis=0))
        bins = list(np.unique(bins))
        if len(bins) == 2:
            bins.append(np.unique(x[~pd.isna(x)])[1])
    else:
        for value in sorted(np.unique(x[~pd.isna(x)])):
            bins.append(value)

    bins.append(np.inf)

    bad_rates, _ = bin_bad_rate(x, y, bins)

    if (pd.isna(bad_rates[0]["bad_rate"])) and (len(bad_rates) > 2):
        del bins[1]
        bad_rates, _ = bin_bad_rate(x, y, bins)

    if len(y[pd.isna(x)]) > 0:
        na_bad_rate = y[pd.isna(x)].sum() / len(y[pd.isna(x)])
        if len(bad_rates) == 2:
            if na_bad_rate < bad_rates[1]["bad_rate"]:
                x = np.nan_to_num(x, nan=np.amin(x[~pd.isna(x)]) - 1)
                bins = [np.NINF, np.amin(x[~pd.isna(x)])] + bins[1:]
                missing_bin = "first"
            else:
                x = np.nan_to_num(x, nan=np.amax(x[~pd.isna(x)]) + 1)
                bins = bins[:2] + [np.amax(x[~pd.isna(x)]), np.inf]
                missing_bin = "last"
        elif abs(
                na_bad_rate
                - np.mean(
                    [bad_rate["bad_rate"] for bad_rate in bad_rates[: len(bad_rates) // 2]]
                )
        ) < abs(
            na_bad_rate
            - np.mean(
                [bad_rate["bad_rate"] for bad_rate in bad_rates[len(bad_rates) // 2:]]
            )
        ):
            x = np.nan_to_num(x, nan=np.amin(x[~pd.isna(x)]))
            missing_bin = "first"
        else:
            x = np.nan_to_num(x, nan=np.amax(x[~pd.isna(x)]))
            missing_bin = "last"
        bad_rates, _ = bin_bad_rate(x, y, bins)

    if len(bad_rates) <= 2:
        return bad_rates, missing_bin

    while (_mono_flags(bad_rates) is True) and (len(bad_rates) > 2):
        bad_rates, bins = _merge_bins_chi(x, y, bad_rates, bins)

    if len(bad_rates) <= 2:
        return bad_rates, missing_bin

    while (min([bad_rate["pct"] for bad_rate in bad_rates]) <= min_pct_group) and (
            len(bad_rates) > 2
    ):
        bad_rates, bins = _merge_bins_min_pct(x, y, bad_rates, bins)

    if len(bad_rates) <= 2:
        return bad_rates, missing_bin

    while (_check_diff_woe(bad_rates, diff_woe_threshold) is not None) and (
            len(bad_rates) > 2
    ):
        idx = _check_diff_woe(bad_rates, diff_woe_threshold) + 1
        del bins[idx]
        bad_rates, _ = bin_bad_rate(x, y, bins)

    return bad_rates, missing_bin

def refit_woe_dict(x: np.ndarray, y: np.ndarray, bins: List, type_feature: str, missing_bin: str) -> List[Dict]:
    cat = True if type_feature == "cat" else False

    if cat:
        try:
            x = x.astype(float)
            x[pd.isna(x)] = -1
        except ValueError:
            x = x.astype(object)
            x[pd.isna(x)] = "Missing"
    else:
        if missing_bin == "first":
            x = np.nan_to_num(x, nan=np.amin(x[~pd.isna(x)]) - 1)
        elif missing_bin == "last":
            x = np.nan_to_num(x, nan=np.amax(x[~pd.isna(x)]) + 1)
        bins = list(set(item for sublist in bins for item in sublist))

    bad_rates, _ = bin_bad_rate(x, y, bins, cat=cat)
    return bad_rates

max_issues_repo_name: kiraplenkin/scorecard

max_forks_repo_name: kiraplenkin/scorecard

reason: import numpy,from scipy


text:
#!/usr/bin/python
# -*- coding: utf-8 -*-
import numpy as np
from datetime import datetime

class Star:
    def __init__(self, object_name, m, color=(1, 1, 0, 1)):
        self.name = object_name
        self.m = m  # масса
        self.X = [0]
        self.Y = [0]
        self.Z = [0]
        self.color = color
        # pos = gl.GLScatterPlotItem(pos=array([0, 0, 0]), color=color, size=10)
        # pos.setGLOptions('translucent')
        # plot_wid.addItem(pos)

class Body:
    def __init__(self, object_name, major, m, a, e, i, w, O=0, M=0, at=0, JD=2451545.0, color=(.5, .5, .5, 1)):
        """
        :param str object_name:     name, название
        :param tuple color:         color point, цвет точки
        :param JD:                  julian date, юлианская дата 2451545.0
        :param major:               center of mass, центр масс
        :param float m:             mass, масса
        :param float at:            наклон оси
        :param float a:             большая полуось КМ
        :param float e:             ексцентриситет ε e=c/a
        :param float i:             отклонение°
        :param float w:             аргумент перицентра° ω
        :param float O:             долгота восходящего узла° Ω N
        :param float M:             средняя аномалия° M=E-e*sin(E)
        ∂ φ
        """
        self.name = object_name
        self.color = color
        self.major = major
        self.m = m
        self.at = at
        self.JD = JD
        self.orbit = True
        self.guide = True
        self.size = None
        self.width = None
        k = G * (self.major.m + m)  # µ гравитационный параметр
        n = np.sqrt(k / a ** 3)  # среднее движение
        self.T = 2 * np.pi / n  # период обращения sqrt(((4 * pi**2)/(G * (SUN.M + m))) * a**3) Кеплер 3
        x, y, z = [], [], []
        E = np.radians(M)
        self.age = int(999 / 365 * (datetime.now() - get_g_d(JD)).days)
        for count in range(self.age):
            while abs((M + e * np.sin(E)) - E) > 0.00001:  # последовательные приближения для эксцентрической аномалии
                E = M + e * np.sin(E)
            M += n * 1  # M = n(t−t0)+M0
            r = a * (1 - e * np.cos(E))  # радиус-вектор
            sin_v = (np.sqrt(1 - e ** 2) * np.sin(E)) / (1 - e * np.cos(E))
            cos_v = (np.cos(E) - e) / (1 - e * np.cos(E))
            sin_u = np.sin(np.radians(w)) * cos_v + np.cos(np.radians(w)) * sin_v
            cos_u = np.cos(np.radians(w)) * cos_v - np.sin(np.radians(w)) * sin_v
            x.append(r * (cos_u * np.cos(np.radians(O)) - sin_u * np.sin(np.radians(O)) * np.cos(np.radians(i))))
            y.append(r * (cos_u * np.sin(np.radians(O)) + sin_u * np.cos(np.radians(O)) * np.cos(np.radians(i))))
            z.append(r * (sin_u * np.sin(np.radians(i))))
            # V1 = sqrt(r / p) * e * sinv
            # V2 = sqrt(r / p) * (1 + e * cosv)
        self.X = list(reversed(x)) if self.at > 90 else x
        self.Y = list(reversed(y)) if self.at > 90 else y
        self.Z = list(reversed(z)) if self.at > 90 else z
        # self.X = x
        # self.Y = y
        # self.Z = z
        # F = G * SUN.M * self.m / r ** 2  # сила гравитационного притяжения
        # p = a * (1 - e ** 2)  # фокальный параметр
        # b = sqrt(a * p)  # малая полуось
        # Rper = (1 - e) * a  # радиус перегелия
        # Rafe = (1 + e) * a  # радиус афелия
        # φ = (24 * pi**3 * a**2) / (T**2 * C**2 * (1 - e**2))
        # φ = (6 * pi * G * SUN.M) / (C**2 * a * (1 - e**2))

        if self.m > 1e25:
            self.outer_planets()
        elif self.m > 1e23:
            self.planet()
        elif self.m > 1e20:
            self.dwarf_planet()
        else:
            self.small_body()

    def star(self):
        pass

    def planet(self, orbit=True, guide=True, size=8, width=1):
        return self.paint(orbit=orbit, guide=guide, size=size, width=width)

    def outer_planets(self, orbit=True, guide=True, size=10, width=1):
        return self.paint(orbit=orbit, guide=guide, size=size, width=width)

    def dwarf_planet(self, orbit=True, guide=True, size=6, width=1):
        return self.paint(orbit=orbit, guide=guide, size=size, width=width)

    def small_body(self, orbit=False, guide=False, size=1.5, width=.25):
        return self.paint(orbit=orbit, guide=guide, size=size, width=width)

    def satellite(self, orbit=False, guide=False, size=3, width=1):
        pass

    def paint(self, orbit=True, guide=True, size=4.0, width=1.0):
        self.orbit = orbit
        self.guide = guide
        self.size = size
        self.width = width

def get_g_d(j_d):
    a = int(j_d + 32044)
    b = int((4 * a + 3) / 146097)
    c = a - int((146097 * b) / 4)
    d = int((4 * c + 3) / 1461)
    e = c - int((1461 * d) / 4)
    m = int((5 * e + 2) / 153)
    day = e - int((153 * m + 2) / 5) + 1
    month = m + 3 - 12 * int((m / 10))
    year = 100 * b + d - 4800 + int((m / 10))
    return datetime(year, month, day)

def get_j_d(day, month, year):
    a = int((14 - month) / 12)
    b = year + 4800 - a
    c = month + 12 * a - 3
    return day + int((153 * c + 2) / 5) + 365 * b + int(b / 4) - int(b / 100) + int(b / 400) - 32045

G = 6.67408e-11  # граивтационная постоянная м3·с−2·кг−1 или Н·м2·кг−2
au = 149597870.700  # а.е. астрономическая единица
C = 299792458  # скорость света м/с

max_stars_repo_name: ezinall/StellarPy

max_issues_repo_name: ezinall/StellarPy

max_forks_repo_name: ezinall/StellarPy


text:
#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Time: 2021/3/14 0014 15:53
# @Author&Email: COLFLIP&colflip@163.com
# @File: Algorithms3.py
# @Software: PyCharm

# ---------------------------------------------
# # Algorithms3：Normalized spectral clustering according toNg et al.(2002)
# Lsym:=D−1/2LD−1/2=I−D−1/2WD−1/2
# Lrw:=D−1L=I−D−1W
# ---------------------------------------------

import matplotlib.pyplot as plt
import numpy as np
from scipy.linalg import fractional_matrix_power
from sklearn import metrics
from sklearn.cluster import KMeans

def getDistanceMatrix(data):
    n = len(data)
    dist_matrix = np.zeros((n, n))
    for i in range(n):
        for j in range(i + 1, n):
            dist = np.sqrt(np.power(data[i] - data[j], 2).sum())
            dist_matrix[i][j] = dist_matrix[j][i] = dist
    return dist_matrix

def getAdjacencyMatrix(data):
    k = 33
    n = len(data)
    dist_matrix = getDistanceMatrix(data)
    W = np.zeros((n, n))
    for idx, item in enumerate(dist_matrix):
        idx_array = np.argsort(item)
        W[idx][idx_array[1:k + 1]] = 1
    transpW = np.transpose(W)
    return (W + transpW) / 2

def getDegreeMatrix(W):
    D = np.diag(sum(W))
    return D

def getLaplacianMatrix(D, W):
    return D - W

def getEigen(L, D, k):
    """
    获得归一化拉普拉斯矩阵的特征矩阵 Lsym
    :param L:
    :param cluter_num: 聚类数目
    :return:
    """
    DD = fractional_matrix_power(D, -0.5)
    L = DD @ L @ DD
    # print(L)
    eigval, eigvec = np.linalg.eig(L)
    ix = np.argsort(eigval)[0:k]
    return eigvec[:, ix]

def plotRes(data, clusterResult, clusterNum):
    n = len(data)
    scatterColors = ['black', 'blue', 'green', 'yellow', 'red', 'purple', 'orange', 'LightGrey']
    for i in range(clusterNum):
        color = scatterColors[i % len(scatterColors)]
        x1 = [];
        y1 = []
        for j in range(n):
            if clusterResult[j] == i:
                x1.append(data[j, 0])
                y1.append(data[j, 1])
        plt.scatter(x1, y1, c=color, marker='+')
    plt.title("Algorithms3")
    plt.show()

def NSCByNg(data, k):
    W = getAdjacencyMatrix(data)
    D = getDegreeMatrix(W)
    L = getLaplacianMatrix(D, W)
    eigvec = getEigen(L, D, k)

    # tij = uij / (∑ku2ik)1 / 2
    rows = eigvec.shape[0]
    columns = eigvec.shape[1]
    T = np.zeros((rows, columns))
    for i in range(rows):
        for j in range(columns):
            T[i][j] = eigvec[i][j] / np.sqrt(np.sum(eigvec[i] ** 2))

    clf = KMeans(n_clusters=k)
    s = clf.fit(T)
    label = s.labels_
    return label

k = 7
filename = 'Aggregation_cluster=7.txt'
data = np.loadtxt(filename, delimiter='\t')
data = data[0:-1]  # 除了最后一列 最后一列为标签列
data = np.array(data)

label = NSCByNg(data, k)
plotRes(data, label, k)

print(metrics.silhouette_score(data, label))  # 轮廓系数评价
print(metrics.davies_bouldin_score(data, label))  # 戴维森堡丁指数(DBI)评价
print(metrics.calinski_harabasz_score(data, label))  # CH指标评价

max_stars_repo_name: colflip/SpectralClustering

max_issues_repo_name: Colflip/spectralClustering

max_forks_repo_name: Colflip/spectralClustering


text:
"""The main class file representing a kernel."""
# Author: Ioannis Siglidis <y.siglidis@gmail.com>
# License: BSD 3 clause
import collections
import warnings
import copy

import numpy as np

from sklearn.base import BaseEstimator
from sklearn.base import TransformerMixin
from sklearn.exceptions import NotFittedError
from sklearn.utils.validation import check_is_fitted
from sklearn.externals import joblib

from grakel.graph import Graph
from grakel.kernels._c_functions import k_to_ij_triangular
from grakel.kernels._c_functions import k_to_ij_rectangular

# Python 2/3 cross-compatibility import
from six import iteritems
try:
    import itertools.imap as map
except ImportError:
    pass

class Kernel(BaseEstimator, TransformerMixin):
    """A general class for graph kernels.

    At default a kernel is considered as pairwise. Doing so the coder that
    adds a new kernel, possibly only needs to overwrite the attributes:
    `parse_input` and `pairwise_operation` on the new kernel object.

    Parameters
    ----------
    n_jobs : int or None, optional
        Defines the number of jobs of a joblib.Parallel objects needed for parallelization
        or None for direct execution.

    normalize : bool, optional
        Normalize the output of the graph kernel.

    verbose : bool, optional
        Define if messages will be printed on stdout.

    Attributes
    ----------
    X : list
        Stores the input that occurs from parse input, on fit input data.
        Default format of the list objects is `grakel.graph.graph`.

    _graph_format : str
        Stores in which type the graphs will need to be stored.

    _verbose : bool
        Defines if two print arguments on stdout.

    _normalize : bool
        Defines if normalization will be applied on the kernel matrix.

    _valid_parameters : set
        Holds the default valid parameters names for initialization.

    _method_calling : int
        An inside enumeration defines which method calls another method.
            - 1 stands for fit
            - 2 stands for fit_transform
            - 3 stands for transform

    _parallel : sklearn.external.joblib.Parallel or None
        A Parallel initialized object to imply parallelization to kernel execution.
        The use of this object depends on the implementation of each base kernel.

    """

    X = None
    _graph_format = "dictionary"
    _method_calling = 0

    def __init__(self,
                 n_jobs=None,
                 normalize=False,
                 verbose=False):
        """`__init__` for `kernel` object."""
        self.verbose = verbose
        self.n_jobs = n_jobs
        self.normalize = normalize
        self._initialized = dict(n_jobs=False)

    def fit(self, X, y=None):
        """Fit a dataset, for a transformer.

        Parameters
        ----------
        X : iterable
            Each element must be an iterable with at most three features and at
            least one. The first that is obligatory is a valid graph structure
            (adjacency matrix or edge_dictionary) while the second is
            node_labels and the third edge_labels (that fitting the given graph
            format). The train samples.

        y : None
            There is no need of a target in a transformer, yet the pipeline API
            requires this parameter.

        Returns
        -------
        self : object
        Returns self.

        """
        self._is_transformed = False
        self._method_calling = 1

        # Parameter initialization
        self.initialize()

        # Input validation and parsing
        if X is None:
            raise ValueError('`fit` input cannot be None')
        else:
            self.X = self.parse_input(X)

        # Return the transformer
        return self

    def transform(self, X):
        """Calculate the kernel matrix, between given and fitted dataset.

        Parameters
        ----------
        X : iterable
            Each element must be an iterable with at most three features and at
            least one. The first that is obligatory is a valid graph structure
            (adjacency matrix or edge_dictionary) while the second is
            node_labels and the third edge_labels (that fitting the given graph
            format). If None the kernel matrix is calculated upon fit data.
            The test samples.

        Returns
        -------
        K : numpy array, shape = [n_targets, n_input_graphs]
            corresponding to the kernel matrix, a calculation between
            all pairs of graphs between target an features

        """
        self._method_calling = 3
        # Check is fit had been called
        check_is_fitted(self, ['X'])

        # Input validation and parsing
        if X is None:
            raise ValueError('`transform` input cannot be None')
        else:
            Y = self.parse_input(X)

        # Transform - calculate kernel matrix
        km = self._calculate_kernel_matrix(Y)
        self._Y = Y

        # Self transform must appear before the diagonal call on normilization
        self._is_transformed = True
        if self.normalize:
            X_diag, Y_diag = self.diagonal()
            km /= np.sqrt(np.outer(Y_diag, X_diag))
        return km

    def fit_transform(self, X):
        """Fit and transform, on the same dataset.

        Parameters
        ----------
        X : iterable
            Each element must be an iterable with at most three features and at
            least one. The first that is obligatory is a valid graph structure
            (adjacency matrix or edge_dictionary) while the second is
            node_labels and the third edge_labels (that fitting the given graph
            format). If None the kernel matrix is calculated upon fit data.
            The test samples.

        y : None
            There is no need of a target in a transformer, yet the pipeline API
            requires this parameter.

        Returns
        -------
        K : numpy array, shape = [n_targets, n_input_graphs]
            corresponding to the kernel matrix, a calculation between
            all pairs of graphs between target an features

        """
        self._method_calling = 2
        self.fit(X)

        # Transform - calculate kernel matrix
        km = self._calculate_kernel_matrix()

        self._X_diag = np.diagonal(km)
        if self.normalize:
            return km / np.sqrt(np.outer(self._X_diag, self._X_diag))
        else:
            return km

    def _calculate_kernel_matrix(self, Y=None):
        """Calculate the kernel matrix given a target_graph and a kernel.

        Each a matrix is calculated between all elements of Y on the rows and
        all elements of X on the columns.

        Parameters
        ----------
        Y : list, default=None
            A list of graph type objects. If None kernel is calculated between
            X and itself.

        Returns
        -------
        K : numpy array, shape = [n_targets, n_inputs]
            The kernel matrix: a calculation between all pairs of graphs
            between targets and inputs. If Y is None targets and inputs
            are the taken from self.X. Otherwise Y corresponds to targets
            and self.X to inputs.

        """
        if Y is None:
            K = np.zeros(shape=(len(self.X), len(self.X)))
            if self._parallel is None:
                cache = list()
                for (i, x) in enumerate(self.X):
                    K[i, i] = self.pairwise_operation(x, x)
                    for (j, y) in enumerate(cache):
                        K[j, i] = self.pairwise_operation(y, x)
                    cache.append(x)
            else:
                dim = len(self.X)
                n_jobs, nsamples = self._n_jobs, ((dim+1)*(dim))//2

                def kij(k):
                    return k_to_ij_triangular(k, dim)

                split = [iter(((i, j), (self.X[i], self.X[j])) for i, j in
                         map(kij, range(*rg))) for rg in indexes(n_jobs, nsamples)]

                self._parallel(joblib.delayed(assign)(s, K, self.pairwise_operation) for s in split)
            K = np.triu(K) + np.triu(K, 1).T

        else:
            K = np.zeros(shape=(len(Y), len(self.X)))
            if self._parallel is None:
                for (j, y) in enumerate(Y):
                    for (i, x) in enumerate(self.X):
                        K[j, i] = self.pairwise_operation(y, x)
            else:
                dim_X, dim_Y = len(self.X), len(Y)
                n_jobs, nsamples = self._n_jobs, (dim_X * dim_Y)

                def kij(k):
                    return k_to_ij_rectangular(k, dim_X)

                split = [iter(((j, i), (Y[j], self.X[i])) for i, j in
                         map(kij, range(*rg))) for rg in indexes(n_jobs, nsamples)]

                self._parallel(joblib.delayed(assign)(s, K, self.pairwise_operation) for s in split)
        return K

    def diagonal(self):
        """Calculate the kernel matrix diagonal of the fit/transformed data.

        Parameters
        ----------
        None.

        Returns
        -------
        X_diag : np.array
            The diagonal of the kernel matrix between the fitted data.
            This consists of each element calculated with itself.

        Y_diag : np.array
            The diagonal of the kernel matrix, of the transform.
            This consists of each element calculated with itself.

        """
        # Check is fit had been called
        check_is_fitted(self, ['X'])
        try:
            check_is_fitted(self, ['_X_diag'])
        except NotFittedError:
            # Calculate diagonal of X
            self._X_diag = np.empty(shape=(len(self.X),))
            for (i, x) in enumerate(self.X):
                self._X_diag[i] = self.pairwise_operation(x, x)

        try:
            # If transform has happened return both diagonals
            check_is_fitted(self, ['_Y'])
            Y_diag = np.empty(shape=(len(self._Y),))
            for (i, y) in enumerate(self._Y):
                Y_diag[i] = self.pairwise_operation(y, y)

            return self._X_diag, Y_diag
        except NotFittedError:
            # Else just return both X_diag
            return self._X_diag

    def parse_input(self, X):
        """Parse the given input and raise errors if it is invalid.

        Parameters
        ----------
        X : iterable
            For the input to pass the test, we must have:
            Each element must be an iterable with at most three features and at
            least one. The first that is obligatory is a valid graph structure
            (adjacency matrix or edge_dictionary) while the second is
            node_labels and the third edge_labels (that correspond to the given
            graph format). A valid input also consists of graph type objects.

        Returns
        -------
        Xp : list
            List of graph type objects.

        """
        if not isinstance(X, collections.Iterable):
            raise TypeError('input must be an iterable\n')
        else:
            Xp = list()
            for (i, x) in enumerate(iter(X)):
                is_iter = isinstance(x, collections.Iterable)
                if is_iter:
                    x = list(x)
                if is_iter and len(x) in [0, 1, 2, 3]:
                    if len(x) == 0:
                        warnings.warn('Ignoring empty element' +
                                      'on index: '+str(i)+'..')
                        continue
                    elif len(x) == 1:
                        Xp.append(Graph(x[0], {}, {},
                                        self._graph_format))
                    elif len(x) == 2:
                        Xp.append(Graph(x[0], x[1], {}, self._graph_format))
                    else:
                        Xp.append(Graph(x[0], x[1], x[2], self._graph_format))
                elif type(x) is Graph:
                    Xp.append(x)
                else:
                    raise TypeError('Each element of X must have at least ' +
                                    'one and at most 3 elements.\n')
            if len(Xp) == 0:
                raise ValueError('Parsed input is empty.')
            return Xp

    def initialize(self):
        """Initialize all transformer arguments, needing initialisation."""
        if not self._initialized["n_jobs"]:
            if type(self.n_jobs) is not int and self.n_jobs is not None:
                raise ValueError('n_jobs parameter must be an int '
                                 'indicating the number of jobs as in joblib or None')
            elif self.n_jobs is None:
                self._parallel = None
            else:
                self._parallel = joblib.Parallel(n_jobs=self.n_jobs,
                                                 backend="threading",
                                                 pre_dispatch='all')
                self._n_jobs = self._parallel._effective_n_jobs()
            self._initialized["n_jobs"] = True

    def pairwise_operation(self, x, y):
        """Calculate a pairwise kernel between two elements.

        Parameters
        ----------
        x, y : Object
            Objects as occur from parse_input.

        Returns
        -------
        kernel : number
            The kernel value.

        """
        raise NotImplementedError('Pairwise operation is not implemented!')

    def set_params(self, **params):
        """Call the parent method."""
        if len(self._initialized):
            # Copy the parameters
            params = copy.deepcopy(params)

            # Iterate over the parameters
            for key, value in iteritems(params):
                key, delim, sub_key = key.partition('__')
                if delim:
                    if sub_key in self._initialized:
                        self._initialized[sub_key] = False
                elif key in self._initialized:
                    self._initialized[key] = False

        # Set parameters
        super(Kernel, self).set_params(**params)

def indexes(n_jobs, nsamples):
    """Distribute samples accross n_jobs."""
    n_jobs = n_jobs

    if n_jobs >= nsamples:
        for i in range(nsamples):
            yield (i, i+1)
    else:
        ns = nsamples/n_jobs
        start = 0
        for i in range(n_jobs-1):
            end = start + ns
            yield (int(start), int(end))
            start = end
        yield (int(start), nsamples)

def assign(data, K, pairwise_operation):
    """Assign list values of an iterable to a numpy array while calculating a pairwise operation."""
    for d in data:
        K[d[0][0], d[0][1]] = pairwise_operation(d[1][0], d[1][1])


text:
import logging
import json
from pathlib import Path
import pytz
import time
import datetime as dt
import os
import sys
import inspect
from random import shuffle
import traceback
import argparse
import numpy

currentdir = os.path.dirname(os.path.abspath(inspect.getfile(inspect.currentframe())))
parentdir = os.path.dirname(currentdir)
sys.path.insert(0, parentdir)

from Utility.Utils import (
    Utils,
    TradeDirection,
    NotSafeToTradeException,
    MarketClosedException,
)
from Interfaces.IGInterface import IGInterface
from Interfaces.AVInterface import AVInterface
from Strategies.StrategyFactory import StrategyFactory
from Interfaces.Broker import Broker
from Interfaces.MarketProvider import MarketProvider, MarketSource

class TradingBot:
    """
    Class that initialise and hold references of main components like the
    broker interface, the strategy or the epic_ids list
    """

    def __init__(self):
        # Set timezone
        set(pytz.all_timezones_set)

        # Load configuration
        home_path = os.path.expanduser("~")
        config_filepath = "{}/.TradingBot/config/config.json".format(home_path)
        config = self.load_json_file(config_filepath)
        self.read_configuration(config)

        # Read credentials file
        credentials = self.load_json_file(self.credentials_filepath)

        # Setup the global logger
        self.setup_logging()

        # Init trade services and create the broker interface
        self.broker = self.init_trading_services(config, credentials)

        # Create strategy from the factory class
        self.strategy = StrategyFactory(config, self.broker).make_strategy(
            self.active_strategy
        )

        # Create the market provider
        self.market_provider = MarketProvider(config, self.broker)

    def load_json_file(self, filepath):
        """
        Load a JSON formatted file from the given filepath

            - **filepath** The filepath including filename and extension
            - Return a dictionary of the loaded json
        """
        try:
            with open(filepath, "r") as file:
                return json.load(file)
        except IOError:
            logging.error("File not found ({})".format(filepath))
            exit()

    def read_configuration(self, config):
        """
        Read the configuration from the config json
        """
        home = os.path.expanduser("~")
        self.epic_ids_filepath = config["general"]["epic_ids_filepath"].replace(
            "{home}", home
        )
        self.credentials_filepath = config["general"]["credentials_filepath"].replace(
            "{home}", home
        )
        self.debug_log = config["general"]["debug_log"]
        self.enable_log = config["general"]["enable_log"]
        self.log_file = config["general"]["log_file"].replace("{home}", home)
        self.time_zone = config["general"]["time_zone"]
        self.max_account_usable = config["general"]["max_account_usable"]
        self.active_strategy = config["general"]["active_strategy"]

    def setup_logging(self):
        """
        Setup the global logging settings
        """
        # Define the global logging settings
        debugLevel = logging.DEBUG if self.debug_log else logging.INFO
        # If enabled define log file filename with current timestamp
        if self.enable_log:
            log_filename = self.log_file
            time_str = dt.datetime.now().isoformat()
            time_suffix = time_str.replace(":", "_").replace(".", "_")
            log_filename = log_filename.replace("{timestamp}", time_suffix)
            os.makedirs(os.path.dirname(log_filename), exist_ok=True)
            logging.basicConfig(
                filename=log_filename,
                level=debugLevel,
                format="[%(asctime)s] %(levelname)s: %(message)s",
            )
        else:
            logging.basicConfig(
                level=debugLevel, format="[%(asctime)s] %(levelname)s: %(message)s"
            )

    def init_trading_services(self, config, credentials):
        """
        Create instances of the trading services required, such as web interface
        for trading and fetch market data.

            - **config** The configuration json
            - **credentials** The credentials json
            - return: An instance of Broker class initialised
        """
        services = {
            "ig_index": IGInterface(config, credentials),
            "alpha_vantage": AVInterface(credentials["av_api_key"], config),
        }
        return Broker(config, services)

    def start(self):
        """
        Starts the TradingBot main loop
        - process open positions
        - process markets from market source
        - wait for configured wait time
        - start over
        """
        while True:
            try:
                # Process current open positions
                self.process_open_positions()
                # Now process markets from the configured market source
                self.process_market_source()
                # Wait for the next spin before starting over
                self.wait_for_strategy_spin_period()
            except MarketClosedException:
                self.wait_for_next_market_opening()
            except NotSafeToTradeException:
                self.wait_for_strategy_spin_period()
            except StopIteration:
                self.wait_for_strategy_spin_period()
            except Exception as e:
                logging.error("Generic exception caught: {}".format(e))
                logging.debug(traceback.format_exc())
                logging.debug(sys.exc_info()[0])
                continue

    def process_open_positions(self):
        """
        Fetch open positions markets and run the strategy against them closing the
        trades if required
        """
        positions = self.broker.get_open_positions()
        # Do not run until we know the current open positions
        if positions is None:
            logging.warning("Unable to fetch open positions! Will try again...")
            raise RuntimeError("Unable to fetch open positions")
        for epic in [item["market"]["epic"] for item in positions["positions"]]:
            market = self.market_provider.get_market_from_epic(epic)
            self.process_market(market, positions)

    def process_market_source(self):
        """
        Process markets from the configured market source
        """
        while True:
            market = self.market_provider.next()
            positions = self.broker.get_open_positions()
            if positions is None:
                logging.warning("Unable to fetch open positions! Will try again...")
                raise RuntimeError("Unable to fetch open positions")
            self.process_market(market, positions)

    def process_market(self, market, open_positions):
        """Spin the strategy on all the markets"""
        self.safety_checks()
        logging.info("Processing {}".format(market.id))
        try:
            self.strategy.set_open_positions(open_positions)
            trade, limit, stop = self.strategy.run(market)
            self.process_trade(market, trade, limit, stop, open_positions)
        except Exception as e:
            logging.error("Strategy exception caught: {}".format(e))
            logging.debug(traceback.format_exc())
            logging.debug(sys.exc_info()[0])
            return

    def close_open_positions(self):
        """
        Closes all the open positions in the account
        """
        logging.info("Closing all the open positions...")
        if self.broker.close_all_positions():
            logging.info("All the posisions have been closed.")
        else:
            logging.error("Impossible to close all open positions, retry.")

    def wait_for_next_market_opening(self):
        """
        Sleep until the next market opening. Takes into account weekends
        and bank holidays in UK
        """
        seconds = Utils.get_seconds_to_market_opening(dt.datetime.now())
        logging.info(
            "Market is closed! Wait for {0:.2f} hours...".format(seconds / 3600)
        )
        time.sleep(seconds)

    def wait_for_strategy_spin_period(self):
        """
        Sleep for the amount of time configured in the active strategy between each spin
        """
        # Wait for next spin loop as configured in the strategy
        seconds = self.strategy.get_seconds_to_next_spin()
        logging.info("Wait for {0:.2f} seconds before next spin".format(seconds))
        time.sleep(seconds)

    def safety_checks(self):
        """
        Perform some safety checks before running the strategy against the next market

        Return True if the trade can proceed, False otherwise
        """
        percent_used = self.broker.get_account_used_perc()
        if percent_used is None:
            logging.warning(
                "Stop trading because can't fetch percentage of account used"
            )
            raise NotSafeToTradeException()
        if percent_used >= self.max_account_usable:
            logging.warning(
                "Stop trading because {}% of account is used".format(str(percent_used))
            )
            raise NotSafeToTradeException()
        if not Utils.is_market_open(self.time_zone):
            logging.warning("Market is closed: stop processing")
            raise MarketClosedException()

    def process_trade(self, market, direction, limit, stop, open_positions):
        """
        Process a trade checking if it is a "close position" trade or a new trade
        """
        # Perform trade only if required
        if direction is not TradeDirection.NONE:
            if open_positions is not None:
                for item in open_positions["positions"]:
                    # If a same direction trade already exist, don't trade
                    if (
                        item["market"]["epic"] == market.epic
                        and direction.name == item["position"]["direction"]
                    ):
                        logging.info(
                            "There is already an open position for this epic, skip trade"
                        )
                        return
                    # If a trade in opposite direction exist, close the position
                    elif (
                        item["market"]["epic"] == market.epic
                        and direction.name != item["position"]["direction"]
                    ):
                        self.broker.close_position(item)
                        return
                self.broker.trade(market.epic, direction.name, limit, stop)
            else:
                logging.error("Unable to fetch open positions! Avoid trading this epic")

def main():
    # Argument management
    parser = argparse.ArgumentParser()
    parser.add_argument(
        "-c",
        "--close_positions",
        help="Close all the open positions",
        action="store_true",
    )
    args = parser.parse_args()

    if args.close_positions:
        TradingBot().close_open_positions()
    else:
        TradingBot().start()

if __name__ == "__main__":
    main()

max_stars_repo_name: andrebask/TradingBot

max_issues_repo_name: andrebask/TradingBot

max_forks_repo_name: andrebask/TradingBot