import random import time from otree.api import * import json from .utils import calculate_pws, bonacich_centrality, get_neighbors_for_node, calculate_ghost_decision doc = """ Your app description """ class C(BaseConstants): NAME_IN_URL = 'coordination_game' PLAYERS_PER_GROUP = 18 NUM_ROUNDS = 65 class Subsession(BaseSubsession): pass def creating_session(subsession): pass class Group(BaseGroup): bot_decisions = models.LongStringField() # Stores bot choices for the current round: e.g., {"5": 1, "12": 0} class Player(BasePlayer): decision = models.IntegerField() neighbour_data = models.LongStringField() # [{'node': n_id, 'is_bot': True, 'decision': decision}, ...] decision_duration = models.FloatField() timeout_happened = models.BooleanField(initial=False) timeout_count = models.IntegerField(initial=0) is_ghost = models.BooleanField(initial=False) # instruction reading time def get_display_choice(self, d): """Helper to return the symbol chosen for the data export""" if d == 1: return self.participant.symbol_for_one else: return "@" if self.participant.symbol_for_one == "#" else "#" def get_neighbor_data(self): import json group = self.group # 1. Get neighbors of my node neighbor_nodes = json.loads(self.participant.neighbour_node_ids) # 2. Get the mappings group_key = self.group.get_players()[0].participant.code node_map = self.session.vars['group_maps'][group_key] bot_nodes = json.loads(self.participant.bot_nodes) bot_decisions = json.loads(group.bot_decisions) results = [] for n_id in neighbor_nodes: if n_id in bot_nodes: # Handle Bot Logic decision = bot_decisions[str(n_id)] results.append({'node': n_id, 'is_bot': True, 'decision': decision}) else: # Handle Human Logic val = node_map.get(str(n_id)) if val is not None: p_id = int(val) neighbor_player = self.subsession.get_players()[p_id - 1] decision = neighbor_player.decision results.append({ 'node': n_id, 'is_bot': False, 'decision': decision }) return results # PAGES class Decision(Page): form_model = 'player' form_fields = ['decision'] @staticmethod def get_timeout_seconds(player: Player): current_strikes = player.participant.timeout_strikes if current_strikes >= 3: return 1 return player.session.config['timeout_total'] @staticmethod def vars_for_template(player: Player): import json # Symbole holen sym_1 = player.participant.symbol_for_one sym_0 = player.participant.symbol_for_zero # Netzwerk-Größe für die Formel neighbor_ids = json.loads(player.participant.neighbour_node_ids) num_neighbors = len(neighbor_ids) # Parameter aus der Config (mit Fallback-Werten) bp0 = player.session.config.get('bp0', 100) bp1 = player.session.config.get('bp1', 50) ep = player.session.config.get('ep', 150) # Basispunkte je nach Phase if not player.participant.in_phase_2: base_1 = bp1 base_0 = bp0 else: base_1 = bp0 base_0 = bp0 rnd_position = random.random() < 0.5 if rnd_position: choice_order = [1, 0] else: choice_order = [0, 1] return dict( label_for_1=sym_1, label_for_0=sym_0, base_1=base_1, base_0=base_0, ep=ep, num_neighbors=num_neighbors, choice_order=choice_order, rnd_position=rnd_position ) @staticmethod def before_next_page(player: Player, timeout_happened): import time # participant._last_page_timestamp ist ein interner oTree-Wert start_time = player.participant._last_page_timestamp if start_time: player.decision_duration = time.time() - start_time current_strikes = player.participant.timeout_strikes if timeout_happened and not player.participant.is_finished: player.timeout_happened = True current_strikes += 1 player.participant.timeout_strikes = current_strikes player.timeout_count = current_strikes if current_strikes >= 3: player.is_ghost = True # if a player did not make a decision the bot logic will take over if player.decision is None: player.decision = calculate_ghost_decision(player) if not timeout_happened: player.participant.timeout_strikes = 0 @staticmethod def is_displayed(player: Player): strikes = player.participant.timeout_strikes if strikes >= 3: return False # Only show if the participant hasn't reached their specific end round return not player.participant.is_finished class ResultsWaitPage(WaitPage): allow_skip = True @staticmethod def after_all_players_arrive(group: Group): from .utils import calculate_bot_decisions # before bot decisions are calculated ghost decisions are calculated for p in group.get_players(): if p.participant.is_finished: continue strikes = p.participant.timeout_strikes if strikes >= 3: p.is_ghost = True # Automatic selection of Ghosts if p.field_maybe_none('decision') is None: p.decision = calculate_ghost_decision(p) elif p.decision is None: p.decision = calculate_ghost_decision(p) # Run bot logic for their decisions for THIS round calculate_bot_decisions(group) for p in group.get_players(): if p.participant.is_finished: continue neighbor_data = p.get_neighbor_data() if p.decision is None or p.is_ghost: p.payoff = 0 else: # calculate payoff for all players that commited a decision num_neighbors = len(neighbor_data) # Count matches with the player's decision matches = sum(1 for n in neighbor_data if n['decision'] == p.decision) bp0 = p.session.config.get('bp0', 100) bp1 = p.session.config.get('bp1', 50) ep = p.session.config.get('ep', 150) # Basic Logic Points if not p.participant.in_phase_2: base_points = bp1 if p.decision == 1 else bp0 else: base_points = bp0 # Bonus for coordination coordination_bonus = ep * (matches / num_neighbors) p.payoff = base_points + coordination_bonus p.neighbour_data = json.dumps(neighbor_data) @staticmethod def is_displayed(player: Player): strikes = player.participant.timeout_strikes if strikes >= 3: return False # Only show if the participant hasn't reached their specific end round return not player.participant.is_finished class Results(Page): @staticmethod def get_timeout_seconds(player: Player): current_strikes = player.participant.timeout_strikes if current_strikes >= 3: return 1 return player.session.config['timeout_total'] @staticmethod def is_displayed(player: Player): strikes = player.participant.timeout_strikes if strikes >= 3: return False # Only show if the participant hasn't reached their specific end round return not player.participant.is_finished @staticmethod def vars_for_template(player: Player): neighbor_data = player.get_neighbor_data() num_neighbors = len(neighbor_data) ep = player.session.config.get('ep', 150) # Zähle Übereinstimmungen count_1 = sum(1 for n in neighbor_data if n['decision'] == 1) count_0 = num_neighbors - count_1 # Welches Symbol hat der Spieler gewählt? sym_1 = player.participant.symbol_for_one sym_0 = player.participant.symbol_for_zero my_symbol = sym_1 if player.decision == 1 else sym_0 # Wie viele Nachbarn haben das GLEICHE gewählt? matches = count_1 if player.decision == 1 else count_0 total_payoff = int(player.payoff) # Da wir base_points nicht gespeichert haben, rechnen wir sie kurz zurück # oder übergeben sie einfach (wie oben in der Logik) bp0 = player.session.config.get('bp0', 100) bp1 = player.session.config.get('bp1', 50) if not player.participant.in_phase_2: base_points = bp1 if player.decision == 1 else bp0 else: base_points = bp0 return dict( neighbor_data=neighbor_data, count_1=count_1, count_0=count_0, matches=matches, num_neighbors=num_neighbors, sym_1=sym_1, sym_0=sym_0, ep=ep, my_symbol=my_symbol, my_decision=player.decision, base_points_payoff=base_points, coordination_payoff=total_payoff- base_points, total_payoff= total_payoff, ) @staticmethod def before_next_page(player: Player, timeout_happened): import json group = player.group current_time = time.time() choices = [p.decision for p in group.get_players()] bot_data = json.loads(group.bot_decisions) choices.extend(bot_data.values()) threshold = player.session.config['consensus_threshold'] end_round = player.participant.experiment_end_round # Wir führen die Prüfung nur durch, wenn wir noch in Phase 1 sind if not player.participant.in_phase_2: max_perc = max(choices.count(1), choices.count(0)) / len(choices) min_p1 = player.session.config['min_rounds_p1'] max_p1 = player.session.config['max_rounds_p1'] # Check for phase change if (player.round_number >= min_p1 and max_perc >= threshold) or (player.round_number == max_p1): p1_end = player.round_number end_r = player.round_number + player.session.config['phase2_length'] est_opt = 1 if choices.count(1) > choices.count(0) else 0 # Setze Phase 2 für ALLE Spieler der Gruppe (wichtig!) for p in group.get_players(): p.participant.in_phase_2 = True p.participant.experiment_end_round = end_r p.participant.established_option = est_opt p.participant.p2_start_time = current_time p.participant.phase_1_end_round = p1_end else: # check for tipping success alternative_perc = 1 - (choices.count(player.participant.established_option) / len(choices)) p1_end = player.participant.phase_1_end_round if end_round and alternative_perc >= threshold: if not player.participant.social_tipping_success: player.participant.rounds_till_success = player.round_number - p1_end player.participant.social_tipping_success = True # Check for the final conclusion of the experiment (Phase 2 complete) if end_round and player.round_number >= end_round: player.participant.is_finished = True player.participant.last_round_decision = player.decision player.participant.p2_end_time = current_time page_sequence = [Decision, ResultsWaitPage, Results]