Coverage for adrs/oms/calculation.py: 72%
95 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-29 05:07 +0800
1import math
2import random
3import logging
5import numpy as np
7from decimal import Decimal, getcontext
9from cybotrade.models import OrderSide
11logger = logging.getLogger(__name__)
13getcontext().prec = 28
16class Calculate:
17 @staticmethod
18 def decimals_sum_all(array):
19 decimal_array = [Decimal(str(f)) for f in array]
20 return float(sum(decimal_array))
22 @staticmethod
23 def round_with_precision(number: Decimal, decimal_places: int) -> Decimal:
24 """
25 Truncate a decimal to the specified number of decimal places.
27 Args:
28 number (Decimal | float): The number to truncate.
29 decimal_places (int): The number of decimal places to keep.
31 Returns:
32 Decimal: The truncated number.
33 """
34 number = Decimal(str(number)) if isinstance(number, float) else number
36 factor = Decimal(str(10)) ** Decimal(str(decimal_places))
37 try:
38 return math.trunc(factor * number) / factor
39 except Exception as e:
40 logger.error(
41 f"[ROUND_WITH_PRECISION] Failed to round with precision due to: {e}"
42 )
43 return Decimal("0.0")
45 @staticmethod
46 def get_max_replace_price(
47 entry_price: Decimal, threshold: Decimal, side: OrderSide, price_precision: int
48 ):
49 coeff = Decimal(str(1 + threshold / 100))
50 coeff1 = Decimal(str(1 - threshold / 100))
51 if side == OrderSide.BUY:
52 if threshold == str(100.0):
53 max_price = entry_price * Decimal(str(10.0))
54 else:
55 max_price = entry_price * coeff
56 else:
57 if threshold == str(100.0):
58 max_price = entry_price / Decimal(str(10.0))
59 else:
60 max_price = entry_price * coeff1
62 return Calculate.round_with_precision(max_price, price_precision)
64 @staticmethod
65 def generate_random_order_size(
66 sum_total: Decimal,
67 count: int,
68 min_qty: Decimal,
69 precision: int,
70 max_qty: Decimal,
71 ) -> list[Decimal]:
72 random_values: list[Decimal] = []
74 # If total is less than min_qty, just return sum_total as one qty
75 if sum_total < min_qty:
76 random_values.append(Decimal("0.0"))
77 logger.info(f"[RANDOM_QTY] sum_total <= min_qty, qty: {random_values}")
78 return random_values
80 # Calculate maximum possible count given sum_total and min_qty
81 max_possible_count = int(
82 (sum_total / min_qty).to_integral_value(rounding="ROUND_FLOOR")
83 )
84 actual_count = min(count, max_possible_count)
86 if actual_count == 0:
87 # sum_total smaller than min_qty, return as one qty
88 random_values.append(Decimal("0.0"))
89 return random_values
91 remaining_total = sum_total
93 for i in range(actual_count - 1):
94 # Maximum for current allocation ensuring enough left for min_qty for remaining
95 max_for_this = min(
96 max_qty,
97 (remaining_total - min_qty * (actual_count - i - 1)).copy_abs(),
98 )
100 # Minimum for current allocation is min_qty or zero if remaining_total < min_qty
101 min_for_this = min_qty if remaining_total >= min_qty else remaining_total
103 # If max_for_this < min_for_this, set max_for_this = min_for_this to avoid invalid uniform range
104 if max_for_this < min_for_this:
105 max_for_this = min_for_this
107 random_float = np.random.uniform(float(min_for_this), float(max_for_this))
108 qty = Decimal(str(random_float)).quantize(Decimal(f"1e-{precision}"))
110 # Clamp qty between 0 and max_qty and not more than remaining_total
111 qty = max(min_qty, min(qty, min(max_qty, remaining_total)))
113 random_values.append(qty)
114 remaining_total -= qty
116 # Avoid negative remaining_total
117 if remaining_total < 0:
118 remaining_total = Decimal("0")
119 break
121 # Append the last qty as the remainder (or zero if negative)
122 last_qty = max(min_qty, remaining_total).quantize(Decimal(f"1e-{precision}"))
123 random_values.append(last_qty)
125 # Remove any zeros if sum_total is very small or rounding causes zeros
126 random_values = [qty for qty in random_values if qty > 0]
128 # Adjust sum to exactly sum_total by tweaking first qty
129 current_sum = sum(random_values)
130 diff = sum_total - current_sum
132 if random_values and diff != 0:
133 adjusted = random_values[0] + diff
134 if adjusted > 0 and adjusted <= max_qty:
135 random_values[0] = adjusted.quantize(Decimal(f"1e-{precision}"))
137 logger.debug(f"[RANDOM_QTY] final random_qty : {random_values}")
139 return random_values
141 @staticmethod
142 def generate_multi_price_level(
143 level: int, tick_size: Decimal, precision: int, price: Decimal, side: OrderSide
144 ) -> list[list[Decimal]]:
145 all_price_level: list[list[Decimal]] = []
146 for i in range(0, level):
147 random_multi = random.choice(
148 [
149 Decimal("0.0001"),
150 Decimal("0.0002"),
151 Decimal("0.0003"),
152 Decimal("0.0004"),
153 Decimal("0.0005"),
154 ]
155 )
156 if side == OrderSide.BUY:
157 raw_price = price * (1 - i * random_multi)
158 aligned_price = math.floor(raw_price / tick_size) * tick_size
159 aligned_price = Calculate.round_with_precision(aligned_price, precision)
160 all_price_level.append([aligned_price, 1 - i * random_multi])
161 else:
162 raw_price = price * (1 + i * random_multi)
163 aligned_price = math.ceil(raw_price / tick_size) * tick_size
164 aligned_price = Calculate.round_with_precision(aligned_price, precision)
165 all_price_level.append([aligned_price, 1 + i * random_multi])
166 all_price_level = sorted(
167 all_price_level, key=lambda x: x[0], reverse=(side == OrderSide.BUY)
168 )
170 return all_price_level
172 @staticmethod
173 def align_price(
174 limit_price: Decimal,
175 side: OrderSide,
176 offset: Decimal,
177 tick_size: Decimal,
178 precision: int,
179 ) -> Decimal:
180 if side == OrderSide.BUY:
181 raw_price = limit_price * (1 - offset)
182 aligned_price = math.floor(raw_price / tick_size) * tick_size
183 else:
184 raw_price = limit_price * (1 + offset)
185 aligned_price = math.ceil(raw_price / tick_size) * tick_size
186 return Calculate.round_with_precision(aligned_price, precision)