45 lines
1.6 KiB
Python
45 lines
1.6 KiB
Python
|
from typing import List, Dict, Any
|
||
|
|
from decimal import Decimal, getcontext, ROUND_DOWN
|
||
|
|
|
||
|
|
getcontext().prec = 50 # 高精度，避免极端值丢失
|
||
|
|
|
||
|
|
def truncate_to_decimals_js(number: float, dots: int) -> float:
|
||
|
|
dec = Decimal(str(number))
|
||
|
|
# 转成字符串，避免科学计数法
|
||
|
|
s = format(dec, "f")
|
||
|
|
if "." not in s:
|
||
|
|
return float(s)
|
||
|
|
integer_part, frac_part = s.split(".")
|
||
|
|
truncated_frac = frac_part[:dots] # 截取小数点后 dots 位
|
||
|
|
truncated_str = f"{integer_part}.{truncated_frac}" if truncated_frac else integer_part
|
||
|
|
return float(truncated_str)
|
||
|
|
|
||
|
|
|
||
|
|
def calculate_shares_weight_js(data: List[Dict[str, Any]]) -> Dict[str, float]:
|
||
|
|
user_weights: Dict[str, Dict[str, Any]] = {}
|
||
|
|
for item in data:
|
||
|
|
user = item.get("user")
|
||
|
|
pool_diff = Decimal(str(item.get("pool_diff", 0)))
|
||
|
|
miner_diff = Decimal(str(item.get("miner_diff", 0)))
|
||
|
|
weight = Decimal("0")
|
||
|
|
if pool_diff != 0:
|
||
|
|
weight = miner_diff / pool_diff
|
||
|
|
if user not in user_weights:
|
||
|
|
user_weights[user] = {"totalWeight": Decimal("0"), "data": []}
|
||
|
|
user_weights[user]["totalWeight"] += weight
|
||
|
|
user_weights[user]["data"].append(item)
|
||
|
|
|
||
|
|
total_weight = sum(u["totalWeight"] for u in user_weights.values())
|
||
|
|
|
||
|
|
if total_weight == 0:
|
||
|
|
return {user: 0.0 for user in user_weights.keys()}
|
||
|
|
|
||
|
|
result: Dict[str, float] = {}
|
||
|
|
for user, v in user_weights.items():
|
||
|
|
ratio = v["totalWeight"] / total_weight
|
||
|
|
result[user] = truncate_to_decimals_js(float(ratio), 10)
|
||
|
|
|
||
|
|
return result
|
||
|
|
|
||
|
|
|
||
|
|
__all__ = ["calculate_shares_weight_js", "truncate_to_decimals_js"]
|