from nullptr.models.marketplace import Marketplace from nullptr.models.jumpgate import Jumpgate from nullptr.models.system import System from nullptr.models.waypoint import Waypoint from dataclasses import dataclass @dataclass class TradeOption: resource: str source: Waypoint dest: Waypoint margin: int dist: int score: float @dataclass class SearchNode: system: System parent: 'SearchNode' def __hash__(self): return hash(self.system.symbol) def path(self): result = [] n = self while n is not None: result.append(n.system) n = n.parent result.reverse() return result def __repr__(self): return self.system.symbol class Analyzer: def __init__(self, store): self.store = store def find_markets(self, resource, sellbuy): for m in self.store.all(Marketplace): if 'sell' in sellbuy and resource in m.imports: yield ('sell', m) elif 'buy' in sellbuy and resource in m.exports: yield ('buy', m) elif 'exchange' in sellbuy and resource in m.exchange: yield ('exchange', m) def find_closest_markets(self, resource, sellbuy, location): if type(location) == str: location = self.store.get(Waypoint, location) mkts = self.find_markets(resource, sellbuy) candidates = [] origin = location.system for typ, m in mkts: system = m.waypoint.system d = origin.distance(system) candidates.append((typ, m, d)) possibles = sorted(candidates, key=lambda m: m[2]) possibles = possibles[:10] results = [] print(len(possibles)) for typ,m,d in possibles: system = m.waypoint.system p = self.find_path(origin, system) if p is None: continue results.append((typ,m,d,len(p))) return results def solve_tsp(self, waypoints): # todo actually try to solve it return waypoints def get_jumpgate(self, system): gates = self.store.all_members(system, Jumpgate) return next(gates, None) def find_path(self, orig, to, depth=100, seen=None): if depth < 1: return None if seen is None: seen = set() if type(orig) == System: orig = set([SearchNode(orig,None)]) result = [n for n in orig if n.system==to] if len(result) > 0: return result[0].path() dest = set() for o in orig: jg = self.get_jumpgate(o.system) if jg is None: continue for s in jg.systems: if s in seen: continue seen.add(s) dest.add(SearchNode(s, o)) if len(dest) == 0: return None return self.find_path(dest, to, depth-1, seen) def prices(self, system): prices = {} for m in self.store.all_members(system, Marketplace): for p in m.prices.values(): r = p['symbol'] if not r in prices: prices[r] = [] prices[r].append({ 'wp': m.waypoint, 'buy': p['buy'], 'sell': p['sell'] }) return prices def find_trade(self, system): prices = self.prices(system) best = None for resource, markets in prices.items(): source = sorted(markets, key=lambda x: x['buy'])[0] dest = sorted(markets, key=lambda x: x['sell'])[-1] margin = dest['sell'] -source['buy'] if margin < 0: continue dist = source['wp'].distance(dest['wp']) dist = max(dist, 0.0001) score = margin / dist o = TradeOption(resource, source['wp'], dest['wp'], margin, dist, score) if best is None or best.score < o.score: best = o return best