KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use d_S method in fMBT

Best Python code snippet using fMBT_python

order_book.py

Source:order_book.py

1from __future__ import annotations2from dataclasses import dataclass, replace3from typing import Sequence, Tuple, Optional, List4@dataclass(frozen=True)5class DollarsAndShares:6    dollars: float7    shares: int8PriceSizePairs = Sequence[DollarsAndShares]9@dataclass(frozen=True)10class OrderBook:11    descending_bids: PriceSizePairs12    ascending_asks: PriceSizePairs13    def bid_price(self) -> float:14        return self.descending_bids[0].dollars15    def ask_price(self) -> float:16        return self.ascending_asks[0].dollars17    def mid_price(self) -> float:18        return (self.bid_price() + self.ask_price()) / 219    def bid_ask_spread(self) -> float:20        return self.ask_price() - self.bid_price()21    def market_depth(self) -> float:22        return self.ascending_asks[-1].dollars - \23            self.descending_bids[-1].dollars24    @staticmethod25    def eat_book(26        ps_pairs: PriceSizePairs,27        shares: int28    ) -> Tuple[DollarsAndShares, PriceSizePairs]:29        '''30        Returned DollarsAndShares represents the pair of31        dollars transacted and the number of shares transacted32        on ps_pairs (with number of shares transacted being less33        than or equal to the input shares).34        Returned PriceSizePairs represents the remainder of the35        ps_pairs after the transacted number of shares have eaten into36        the input ps_pairs.37        '''38        rem_shares: int = shares39        dollars: float = 0.40        for i, d_s in enumerate(ps_pairs):41            this_price: float = d_s.dollars42            this_shares: int = d_s.shares43            dollars += this_price * min(rem_shares, this_shares)44            if rem_shares < this_shares:45                return (46                    DollarsAndShares(dollars=dollars, shares=shares),47                    [DollarsAndShares(48                        dollars=this_price,49                        shares=this_shares - rem_shares50                    )] + list(ps_pairs[i+1:])51                )52            else:53                rem_shares -= this_shares54        return (55            DollarsAndShares(dollars=dollars, shares=shares - rem_shares),56            []57        )58    def sell_limit_order(self, price: float, shares: int) -> \59            Tuple[DollarsAndShares, OrderBook]:60        index: Optional[int] = next((i for i, d_s61                                     in enumerate(self.descending_bids)62                                     if d_s.dollars < price), None)63        eligible_bids: PriceSizePairs = self.descending_bids \64            if index is None else self.descending_bids[:index]65        ineligible_bids: PriceSizePairs = [] if index is None else \66            self.descending_bids[index:]67        d_s, rem_bids = OrderBook.eat_book(eligible_bids, shares)68        new_bids: PriceSizePairs = list(rem_bids) + list(ineligible_bids)69        rem_shares: int = shares - d_s.shares70        if rem_shares > 0:71            new_asks: List[DollarsAndShares] = list(self.ascending_asks)72            index1: Optional[int] = next((i for i, d_s73                                          in enumerate(new_asks)74                                          if d_s.dollars >= price), None)75            if index1 is None:76                new_asks.append(DollarsAndShares(77                    dollars=price,78                    shares=rem_shares79                ))80            elif new_asks[index1].dollars != price:81                new_asks.insert(index1, DollarsAndShares(82                    dollars=price,83                    shares=rem_shares84                ))85            else:86                new_asks[index1] = DollarsAndShares(87                    dollars=price,88                    shares=new_asks[index1].shares + rem_shares89                )90            return d_s, OrderBook(91                ascending_asks=new_asks,92                descending_bids=new_bids93            )94        else:95            return d_s, replace(96                self,97                descending_bids=new_bids98            )99    def sell_market_order(100        self,101        shares: int102    ) -> Tuple[DollarsAndShares, OrderBook]:103        d_s, rem_bids = OrderBook.eat_book(104            self.descending_bids,105            shares106        )107        return (d_s, replace(self, descending_bids=rem_bids))108    def buy_limit_order(self, price: float, shares: int) -> \109            Tuple[DollarsAndShares, OrderBook]:110        index: Optional[int] = next((i for i, d_s111                                     in enumerate(self.ascending_asks)112                                     if d_s.dollars > price), None)113        eligible_asks: PriceSizePairs = self.ascending_asks \114            if index is None else self.ascending_asks[:index]115        ineligible_asks: PriceSizePairs = [] if index is None else \116            self.ascending_asks[index:]117        d_s, rem_asks = OrderBook.eat_book(eligible_asks, shares)118        new_asks: PriceSizePairs = list(rem_asks) + list(ineligible_asks)119        rem_shares: int = shares - d_s.shares120        if rem_shares > 0:121            new_bids: List[DollarsAndShares] = list(self.descending_bids)122            index1: Optional[int] = next((i for i, d_s123                                          in enumerate(new_bids)124                                          if d_s.dollars <= price), None)125            if index1 is None:126                new_bids.append(DollarsAndShares(127                    dollars=price,128                    shares=rem_shares129                ))130            elif new_bids[index1].dollars != price:131                new_bids.insert(index1, DollarsAndShares(132                    dollars=price,133                    shares=rem_shares134                ))135            else:136                new_bids[index1] = DollarsAndShares(137                    dollars=price,138                    shares=new_bids[index1].shares + rem_shares139                )140            return d_s, replace(141                self,142                ascending_asks=new_asks,143                descending_bids=new_bids144            )145        else:146            return d_s, replace(147                self,148                ascending_asks=new_asks149            )150    def buy_market_order(151        self,152        shares: int153    ) -> Tuple[DollarsAndShares, OrderBook]:154        d_s, rem_asks = OrderBook.eat_book(155            self.ascending_asks,156            shares157        )158        return (d_s, replace(self, ascending_asks=rem_asks))159    def pretty_print_order_book(self) -> None:160        from pprint import pprint161        print()162        print("Bids")163        pprint(self.descending_bids)164        print()165        print("Asks")166        print()167        pprint(self.ascending_asks)168        print()169    def display_order_book(self) -> None:170        import matplotlib.pyplot as plt171        bid_prices = [d_s.dollars for d_s in self.descending_bids]172        bid_shares = [d_s.shares for d_s in self.descending_bids]173        if self.descending_bids:174            plt.bar(bid_prices, bid_shares, color='blue')175        ask_prices = [d_s.dollars for d_s in self.ascending_asks]176        ask_shares = [d_s.shares for d_s in self.ascending_asks]177        if self.ascending_asks:178            plt.bar(ask_prices, ask_shares, color='red')179        all_prices = sorted(bid_prices + ask_prices)180        all_ticks = ["%d" % x for x in all_prices]181        plt.xticks(all_prices, all_ticks)182        plt.grid(axis='y')183        plt.xlabel("Prices")184        plt.ylabel("Number of Shares")185        plt.title("Order Book")186        # plt.xticks(x_pos, x)187        plt.show()188if __name__ == '__main__':189    from numpy.random import poisson190    bids: PriceSizePairs = [DollarsAndShares(191        dollars=x,192        shares=poisson(100. - (100 - x) * 10)193    ) for x in range(100, 90, -1)]194    asks: PriceSizePairs = [DollarsAndShares(195        dollars=x,196        shares=poisson(100. - (x - 105) * 10)197    ) for x in range(105, 115, 1)]198    ob0: OrderBook = OrderBook(descending_bids=bids, ascending_asks=asks)199    ob0.pretty_print_order_book()200    ob0.display_order_book()201    print("Sell Limit Order of (107, 40)")202    print()203    d_s1, ob1 = ob0.sell_limit_order(107, 40)204    proceeds1: float = d_s1.dollars205    shares_sold1: int = d_s1.shares206    print(f"Sales Proceeds = {proceeds1:.2f}, Shares Sold = {shares_sold1:d}")207    ob1.pretty_print_order_book()208    ob1.display_order_book()209    print("Sell Market Order of 120")210    print()211    d_s2, ob2 = ob1.sell_market_order(120)212    proceeds2: float = d_s2.dollars213    shares_sold2: int = d_s2.shares214    print(f"Sales Proceeds = {proceeds2:.2f}, Shares Sold = {shares_sold2:d}")215    ob2.pretty_print_order_book()216    ob2.display_order_book()217    print("Buy Limit Order of (100, 80)")218    print()219    d_s3, ob3 = ob2.buy_limit_order(100, 80)220    bill3: float = d_s3.dollars221    shares_bought3: int = d_s3.shares222    print(f"Purchase Bill = {bill3:.2f}, Shares Bought = {shares_bought3:d}")223    ob3.pretty_print_order_book()224    ob3.display_order_book()225    print("Sell Limit Order of (104, 60)")226    print()227    d_s4, ob4 = ob3.sell_limit_order(104, 60)228    proceeds4: float = d_s4.dollars229    shares_sold4: int = d_s4.shares230    print(f"Sales Proceeds = {proceeds4:.2f}, Shares Sold = {shares_sold4:d}")231    ob4.pretty_print_order_book()232    ob4.display_order_book()233    print("Buy Market Order of 150")234    print()235    d_s5, ob5 = ob4.buy_market_order(150)236    bill5: float = d_s5.dollars237    shares_bought5: int = d_s5.shares238    print(f"Purchase Bill = {bill5:.2f}, Shares Bought = {shares_bought5:d}")239    ob5.pretty_print_order_book()...

p_decay_isolated.py

Source:p_decay_isolated.py

1import numpy as np2import pandas as pd3import matplotlib.pyplot as plt4import seaborn as sns;5from mpl_toolkits.mplot3d import Axes3D6sns.set()7from scipy.integrate import solve_ivp8from itertools import product9from multiprocessing import Pool10from sklearn.preprocessing import PolynomialFeatures11from sklearn.linear_model import LinearRegression12from scipy.optimize import curve_fit13# %%14def tau_p(p_max, d_s=15, eps=1e-4):15    return d_s / (np.log(p_max) - np.log(eps))16def p_t(t, p_max, **kwargs):17    p = p_max * np.exp(-t / tau_p(p_max, **kwargs))18    return p19def dn_dt(t, n, p_max, **kwargs):20    r_t = p_t(t, p_max, **kwargs) * n21    return -r_t22def n(time_bins, p_max, init_n=20, **kwargs):23    integrated = solve_ivp(fun=lambda t, y: dn_dt(t, y, p_max, **kwargs),24                           t_span=(0, time_bins[-1]),25                           y0=(init_n,),26                           t_eval=time_bins)27    return integrated.y.squeeze()28def p_t_wrong(t, d_s=15, d_lr=9, n_max=20, **kwargs):29    # Sadly, however fancy this function may be it is simply in correct because it operates30    # On the assumption that n is fixed at all time points to arrive at the expected value31    dummy_p_max = 132    tau = tau_p(dummy_p_max, d_s, **kwargs)33    p = np.exp(-t / tau)34    p_sum_lr = tau - (tau * np.exp(-d_lr / tau))35    p = p / p_sum_lr * n_max36    return p37def dp_dt():38    pass39def dn_dt_wrong(t, n, **kwargs):40    r_t = p_t(t, **kwargs) * n41    return -r_t42def n_wrong(time_bins, init_n=20, **kwargs):43    integrated = solve_ivp(fun=lambda t, y: dn_dt(t, y, **kwargs),44                           t_span=(0, time_bins[-1]),45                           y0=(init_n,),46                           t_eval=time_bins)47    return integrated.y.squeeze()48# %% helper functions49def make_t_data(times, d_s, p_max, n_max, **kwargs):50    # Making sure our input variables can be interpreted as arrays if they are not already51    d_s_arr = np.r_[d_s]52    p_max_arr = np.r_[p_max]53    n_max_arr = np.r_[n_max]54    conditions = [55        {'d_s': d_s, 'p_max': p_max, 'n_max': n_max} for d_s, p_max, n_max in product(d_s_arr, p_max_arr, n_max_arr)]56    results = []57    for params in conditions:58        current_result = params.copy()59        param_dict = {key: params[key] for key in params if key != 'n_max'}60        current_p = p_t(times, **param_dict, **kwargs)61        current_n = n(times, init_n=params['n_max'], **param_dict, **kwargs)62        current_result.update({63            't': times,64            'p': current_p,65            'n': current_n,66        })67        results.append(pd.DataFrame(current_result))68    results = pd.concat(results)69    results['r'] = results.p * results.n70    return results71def _make_for_p_params(params, pmax_values, dt, **kwargs):72    times = np.arange(0 + dt, params['t'] + dt, dt)73    current_result = params.copy()74    current_p = p_t(params['t'], np.r_[pmax_values], d_s=params['d_s'], **kwargs)75    current_n = []76    for p_max in pmax_values:77        current_n.append(78            n(times, init_n=params['n_max'], p_max=p_max, d_s=params['d_s'], **kwargs)[-1])79    current_result.update({80        'p_max': pmax_values,81        'p': current_p,82        'n': current_n,83        't': [params['t']] * pmax_values.shape[0]84    })85    return pd.DataFrame(current_result)86def make_p_data(pmax_values, t, d_s, n_max, dt=0.01, **kwargs):87    # Making sure our input variables can be interpreted as arrays if they are not already88    t_arr = np.r_[t]89    d_s_arr = np.r_[d_s]90    n_max_arr = np.r_[n_max]91    conditions = [92        {'t': t, 'd_s': d_s, 'n_max': n_max} for t, d_s, n_max in product(t_arr, d_s_arr, n_max_arr)]93    # results = []94    # for params in conditions:95    #     results.append(_make_for_p_params(params, pmax_values, dt, **kwargs))96    with Pool(8) as P:97        results = P.starmap(_make_for_p_params, ((c, pmax_values, dt) for c in conditions))98    results = pd.concat(results)99    return results100def make_combo_data(times, combos, n_max, **kwargs):101    results = []102    for i, params in combos.iterrows():103        current_result = dict(params)104        current_result['n_max'] = n_max105        current_p = p_t(times, p_max=params['p_max'], d_s=params['d_s'], **kwargs)106        current_n = n(times, init_n=n_max, p_max=params['p_max'], d_s=params['d_s'], **kwargs)107        current_result.update({108            't': times,109            'p': current_p,110            'n': current_n,111        })112        results.append(pd.DataFrame(current_result))113    results = pd.concat(results)114    results['r'] = results.p * results.n115    return results116def plot_data(data, x='t', n_max=np.inf):117    pmax_values = np.unique(data.p_max)118    ds_values = np.unique(data.d_s)119    t_values = np.unique(data.t)120    if x == 't':121        ys = ['p', 'r', 'n']122        fig_shape = (len(ys), len(pmax_values))123        # fig, axarr = plt.subplots(len(ys), len(pmax_values), sharex=True, sharey='row')124        var_values = pmax_values125        var_name = 'p_max'126        hue = 'd_s'127    if (x == 'p_max') or (x == 'p'):128        x = 'p_max'129        ys = ['n', 'p']130        fig_shape = (len(ys), len(ds_values))131        # fig, axarr = plt.subplots(len(ys), len(t_values), sharex=True, sharey='row')132        var_values = ds_values133        var_name = 'd_s'134        hue = 't'135    fig, axarr = plt.subplots(*fig_shape, sharex=True, sharey='row')136    for i, (y, axrow) in enumerate(zip(ys, axarr)):137        if np.ndim(axrow) == 0:138            ax = axrow139            value = var_values140            plot_data = data[(data[var_name] == value[0]) & (data['n'] < n_max)]141            sns.lineplot(x=x, y=y, hue=hue, data=plot_data, ax=ax)142            if hue:143                ax.legend_.remove()144            if i == 0:145                ax.set_title(f'{var_name} = {value[0]}')146        else:147            for value, ax in zip(var_values, axrow):148                plot_data = data[(data[var_name] == value) & (data['n'] < n_max)]149                sns.lineplot(x=x, y=y, hue=hue, data=plot_data, ax=ax)150                if hue:151                    ax.legend_.remove()152                if i == 0:153                    ax.set_title(f'{var_name} = {value}')154    if hue:155        if axarr.ndim == 2:156            axarr[0, 0].legend(np.unique(data[hue]))157        else:158            axarr[0].legend(np.unique(data[hue]))159def plot_combo(combo_data):160    ds_values = np.unique(combo_data.d_s)161    pmax_values = np.unique(combo_data.p_max)162    x = 't'163    ys = ['p', 'r', 'n']164    hue = 'd_s'165    fig_shape = (len(ys), len(pmax_values))166    fig, axarr = plt.subplots(*fig_shape, sharex=True, sharey='row')167    for i, (y, axrow) in enumerate(zip(ys, axarr)):168        for j, (p_max, ax) in enumerate(zip(pmax_values, axrow)):169            plot_data = combo_data[combo_data['p_max'] == p_max]170            sns.lineplot(x=x, y=y, data=plot_data, ax=ax)171            if i==0:172                ax.set_title(f'p_max={p_max} | d_s={np.unique(plot_data.d_s)}')173# %% full dimensional analysis at t=[3, 6, 9]174d_lr = 9175n_max = 100176pmax_range = (0.5, 5)177n_pmax = 250178ds_range = (30, 1000)179n_ds = 250180dt = 0.01  # ms181t_values = [3, 6, 9]182# Setup logarithmically space values183pmax_values = np.linspace(*pmax_range, num=n_pmax)184ds_values = np.linspace(*ds_range, num=n_ds)185# pmax_values = np.logspace(*np.log10(pmax_range), num=n_pmax, endpoint=False)186# ds_values = np.logspace(*np.log10(ds_range), num=n_ds, endpoint=False).round()187# ds_values = np.logspace(np.log10(30), np.log10(1000), 4, endpoint=False).round()188mecha_data = make_p_data(pmax_values=pmax_values, t=t_values,189                     d_s=ds_values, n_max=n_max, dt=dt)190# %%191fig = plt.figure()192selected_data = mecha_data[mecha_data.t == 9]193selected_data['gap_69'] = mecha_data.n[mecha_data.t == 6] - selected_data.n194inds = (selected_data.n <= (n_max  - 20)) & (selected_data.n > (n_max - 20 - 1))195# selected_data = selected_data[inds]196ax = fig.add_subplot(1,2, 1, projection='3d')197ax2 = fig.add_subplot(1,2, 2, projection='3d')198ax.plot_trisurf(selected_data['p_max'], selected_data['d_s'], selected_data['n'], cmap=plt.cm.viridis, linewidth=0.2)199ax.set_title('n at t=9')200ax2.plot_trisurf(selected_data['p_max'], selected_data['d_s'], selected_data['gap_69'], cmap=plt.cm.viridis, linewidth=0.2)201ax2.set_title('6-9 gap')202# %%203plot_duration = 15204d_lr = 9205n_max = 20206dt = 0.01  # ms207times = np.arange(0 + dt, plot_duration + dt, dt)208combos = pd.DataFrame({209    'p_max': [1.55, 0.86, 0.68, 0.55, 0.46],210    'd_s': [25, 45, 80, 140, 400]211})212combo_data = make_combo_data(times, combos, n_max)213plot_combo(combo_data)214## (p_max =0.86, d_s=45)215# %%216d_lr = 9217n_max = 40218pmax_range = (0.01, 1)219n_pmax = 100220# ds_range = (25, 100)221n_ds = 4222dt = 0.01  # ms223t_values = [3, 6, 9]224# Setup logarithmically space values225pmax_values = np.logspace(*np.log10(pmax_range), num=n_pmax, endpoint=False)226ds_values = np.r_[30, 45, 60, 75]227# ds_values = np.logspace(np.log10(30), np.log10(1000), 4, endpoint=False).round()228p_data = make_p_data(pmax_values=pmax_values, t=t_values,229                     d_s=ds_values, n_max=n_max, dt=dt)...

valid_anagram.py

Source:valid_anagram.py

1# https://leetcode.com/explore/interview/card/top-interview-questions-easy/127/strings/882/2################ MY solution using two hash tables ####################3# class Solution:4#     def isAnagram(self, s: str, t: str) -> bool:5#         d_s = {}6#         d_t = {}7        8#         for i in range(len(s)):9#             if s[i] in d_s.keys():10#                 d_s[s[i]] += 111#             else:12#                 d_s[s[i]] = 113                14#         for i in range(len(t)):15#             if t[i] in d_t.keys():16#                 d_t[t[i]] += 117#             else:18#                 d_t[t[i]] = 119        20#         if len(d_s) > len(d_t):21#             outer = d_s22#             inner = d_t23#         else:24#             outer = d_t25#             inner = d_s26        27#         for key, value in outer.items():28#             if key in inner.keys():29#                 if value == inner[key]:30#                     continue31#                 else:32#                     return False33#             else:34#                 return False35            36            37#         return True38############# A more efficient solution using one hash table #################39# class Solution:40#     def isAnagram(self, s: str, t: str) -> bool:41#         d_s = {}42        43#         for i in range(len(s)):44#             if s[i] in d_s.keys():45#                 d_s[s[i]] += 146#             else:47#                 d_s[s[i]] = 148                49#         for i in range(len(t)):50#             if t[i] in d_s.keys():51#                 d_s[t[i]] -= 152#             else:53#                 return False54        55#         for key, value in d_s.items():56#             if value != 0:57#                 return False58            59            ...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.